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

Fix several restore and resume bugs (#1418)

Browse Source 
* try to correct resume messages with missing checkpoint

* prevent creating checkpoints for outdated task waits

* prevent creating checkpoints for outdated batch waits

* use heartbeats to check for and clean up any leftover containers

* lint

* improve exec logging

* improve resume attempt logs

* fix for resuming parents of canceled child runs

* separate SIGTERM from maybe OOM errors

* pretty errors can have magic dashboard links

* prevent uncancellable checkpoints

* simplify task run error code enum export

* grab the last, not the first child run

* Revert "prevent creating checkpoints for outdated batch waits"

This reverts commit f2b5c2ac42.

* Revert "grab the last, not the first child run"

This reverts commit 89ec5c8bfd.

* Revert "prevent creating checkpoints for outdated task waits"

This reverts commit 11066b4e74.

* more logs for resume message handling

* add magic error link comment

* add changeset
This commit is contained in:
nicktrn
2024-10-18 16:46:38 +01:00
committed by GitHub
parent 235ab90c3d
commit 90593adea5
12 changed files with 416 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/many-plants-destroy.md
+5
View File
@@ -0,0 +1,5 @@
---
"@trigger.dev/core": patch
---
SIGTERM detection and prettier errors
apps/coordinator/src/checkpointer.ts
+4 -1
View File
@@ -436,7 +436,10 @@ export class Checkpointer {
this.#logger.error("Error during cleanup", { ...metadata, error });
}
this.#abortControllers.delete(runId);
// Ensure only the current controller is removed
if (this.#abortControllers.get(runId) === controller) {
this.#abortControllers.delete(runId);
}
controller.signal.removeEventListener("abort", onAbort);
};
apps/coordinator/src/exec.ts
+12 -1
View File
@@ -64,7 +64,18 @@ export class Exec {
command,
argsRaw: args,
argsTrimmed,
...output,
globalOpts: {
trimArgs: this.trimArgs,
neverThrow: this.neverThrow,
hasAbortSignal: !!this.abortSignal,
},
localOpts: opts,
stdout: output.stdout,
stderr: output.stderr,
pid: result.pid,
exitCode: result.exitCode,
aborted: result.aborted,
killed: result.killed,
};
if (this.logOutput) {
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx
+28 -6
View File
@@ -1,4 +1,10 @@
import { CheckIcon, ClockIcon, CloudArrowDownIcon, QueueListIcon } from "@heroicons/react/20/solid";
import {
CheckIcon,
ClockIcon,
CloudArrowDownIcon,
EnvelopeIcon,
QueueListIcon,
} from "@heroicons/react/20/solid";
import { Link } from "@remix-run/react";
import { LoaderFunctionArgs } from "@remix-run/server-runtime";
import {
@@ -13,6 +19,7 @@ import { typedjson, useTypedFetcher } from "remix-typedjson";
import { ExitIcon } from "~/assets/icons/ExitIcon";
import { CodeBlock } from "~/components/code/CodeBlock";
import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
import { Feedback } from "~/components/Feedback";
import { Button, LinkButton } from "~/components/primitives/Buttons";
import { Callout } from "~/components/primitives/Callout";
import { DateTime, DateTimeAccurate } from "~/components/primitives/DateTime";
@@ -963,11 +970,26 @@ function RunError({ error }: { error: TaskRunError }) {
<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">{name}</Header3>
{enhancedError.message && <Callout variant="error">{enhancedError.message}</Callout>}
{enhancedError.link && (
<Callout variant="docs" to={enhancedError.link.href}>
{enhancedError.link.name}
</Callout>
)}
{enhancedError.link &&
(enhancedError.link.magic === "CONTACT_FORM" ? (
<Feedback
button={
<Button
variant="tertiary/medium"
LeadingIcon={EnvelopeIcon}
leadingIconClassName="text-blue-400"
fullWidth
textAlignLeft
>
{enhancedError.link.name}
</Button>
}
/>
) : (
<Callout variant="docs" to={enhancedError.link.href}>
{enhancedError.link.name}
</Callout>
))}
{enhancedError.stackTrace && (
<CodeBlock
showCopyButton={false}
apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts
+1 -1
View File
@@ -136,7 +136,7 @@ export class EnvironmentVariablesRepository implements Repository {
try {
for (const variable of values) {
const result = await $transaction(this.prismaClient, async (tx) => {
const result = await $transaction(this.prismaClient, async (tx) => {
const environmentVariable = await tx.environmentVariable.upsert({
where: {
projectId_key: {
apps/webapp/app/v3/handleSocketIo.server.ts
+4 -1
View File
@@ -195,7 +195,10 @@ function createCoordinatorNamespace(io: Server) {
const service = new CreateTaskRunAttemptService();
const { attempt } = await service.call(message.runId, environment, false);
const payload = await sharedQueueTasks.getExecutionPayloadFromAttempt(attempt.id, true);
const payload = await sharedQueueTasks.getExecutionPayloadFromAttempt({
id: attempt.id,
setToExecuting: true,
});
if (!payload) {
logger.error("Failed to retrieve payload after attempt creation", message);
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
+209 -41
View File
@@ -21,7 +21,7 @@ import {
TaskRunStatus,
} from "@trigger.dev/database";
import { z } from "zod";
import { prisma } from "~/db.server";
import { $replica, prisma } from "~/db.server";
import { findEnvironmentById } from "~/models/runtimeEnvironment.server";
import { logger } from "~/services/logger.server";
import { singleton } from "~/utils/singleton";
@@ -43,7 +43,12 @@ import { generateJWTTokenForEnvironment } from "~/services/apiAuth.server";
import { EnvironmentVariable } from "../environmentVariables/repository";
import { machinePresetFromConfig } from "../machinePresets.server";
import { env } from "~/env.server";
import { isFinalAttemptStatus, isFinalRunStatus } from "../taskStatus";
import {
FINAL_ATTEMPT_STATUSES,
FINAL_RUN_STATUSES,
isFinalAttemptStatus,
isFinalRunStatus,
} from "../taskStatus";
import { getMaxDuration } from "../utils/maxDuration";
const WithTraceContext = z.object({
@@ -620,6 +625,9 @@ export class SharedQueueConsumer {
const resumableRun = await prisma.taskRun.findUnique({
where: {
id: message.messageId,
status: {
notIn: FINAL_RUN_STATUSES,
},
},
});
@@ -633,6 +641,14 @@ export class SharedQueueConsumer {
return;
}
if (resumableRun.status !== "EXECUTING") {
logger.warn("Run is not executing, will try to resume anyway", {
queueMessage: message.data,
messageId: message.messageId,
runStatus: resumableRun.status,
});
}
const resumableAttempt = await prisma.taskRunAttempt.findUnique({
where: {
id: messageBody.data.resumableAttemptId,
@@ -718,9 +734,9 @@ export class SharedQueueConsumer {
completions.push(completion);
const executionPayload = await this._tasks.getExecutionPayloadFromAttempt(
completedAttempt.id
);
const executionPayload = await this._tasks.getExecutionPayloadFromAttempt({
id: completedAttempt.id,
});
if (!executionPayload) {
await this.#ackAndDoMoreWork(message.messageId);
@@ -740,7 +756,11 @@ export class SharedQueueConsumer {
executions,
};
logger.debug("Broadcasting RESUME_AFTER_DEPENDENCY_WITH_ACK", { resumeMessage, message });
logger.debug("Broadcasting RESUME_AFTER_DEPENDENCY_WITH_ACK", {
resumeMessage,
message,
resumableRun,
});
// The attempt should still be running so we can broadcast to all coordinators to resume immediately
const responses = await socketIo.coordinatorNamespace
@@ -763,15 +783,91 @@ export class SharedQueueConsumer {
}
const hasSuccess = responses.some((response) => response.success);
if (!hasSuccess) {
logger.warn("RESUME_AFTER_DEPENDENCY_WITH_ACK failed", {
resumeMessage,
responses,
message,
});
await this.#nackAndDoMoreWork(message.messageId, this._options.nextTickInterval, 5_000);
if (hasSuccess) {
this.#doMoreWork();
return;
}
// No coordinator was able to resume the run
logger.warn("RESUME_AFTER_DEPENDENCY_WITH_ACK failed", {
resumeMessage,
responses,
message,
});
// Let's check if the run is frozen
if (resumableRun.status === "WAITING_TO_RESUME") {
logger.debug("RESUME_AFTER_DEPENDENCY_WITH_ACK run is waiting to be restored", {
queueMessage: message.data,
messageId: message.messageId,
});
try {
const restoreService = new RestoreCheckpointService();
const checkpointEvent = await restoreService.getLastCheckpointEventIfUnrestored(
resumableRun.id
);
if (checkpointEvent) {
// The last checkpoint hasn't been restored yet, so restore it
const checkpoint = await restoreService.call({
eventId: checkpointEvent.id,
});
if (!checkpoint) {
logger.debug("RESUME_AFTER_DEPENDENCY_WITH_ACK failed to restore checkpoint", {
queueMessage: message.data,
messageId: message.messageId,
});
await this.#ackAndDoMoreWork(message.messageId);
return;
}
logger.debug("RESUME_AFTER_DEPENDENCY_WITH_ACK restored checkpoint", {
queueMessage: message.data,
messageId: message.messageId,
checkpoint,
});
this.#doMoreWork();
return;
} else {
logger.debug(
"RESUME_AFTER_DEPENDENCY_WITH_ACK run is frozen without last checkpoint event",
{
queueMessage: message.data,
messageId: message.messageId,
}
);
}
} catch (e) {
if (e instanceof Error) {
this._currentSpan?.recordException(e);
} else {
this._currentSpan?.recordException(new Error(String(e)));
}
this._endSpanInNextIteration = true;
await this.#nackAndDoMoreWork(
message.messageId,
this._options.nextTickInterval,
5_000
);
return;
}
}
logger.debug("RESUME_AFTER_DEPENDENCY_WITH_ACK retrying", {
queueMessage: message.data,
messageId: message.messageId,
});
await this.#nackAndDoMoreWork(message.messageId, this._options.nextTickInterval, 5_000);
return;
} catch (e) {
if (e instanceof Error) {
this._currentSpan?.recordException(e);
@@ -896,7 +992,7 @@ class SharedQueueTasks {
where: {
id,
status: {
in: ["COMPLETED", "FAILED"],
in: FINAL_ATTEMPT_STATUSES,
},
},
include: {
@@ -942,11 +1038,17 @@ class SharedQueueTasks {
}
}
async getExecutionPayloadFromAttempt(
id: string,
setToExecuting?: boolean,
isRetrying?: boolean
): Promise<ProdTaskRunExecutionPayload | undefined> {
async getExecutionPayloadFromAttempt({
id,
setToExecuting,
isRetrying,
skipStatusChecks,
}: {
id: string;
setToExecuting?: boolean;
isRetrying?: boolean;
skipStatusChecks?: boolean;
}): Promise<ProdTaskRunExecutionPayload | undefined> {
const attempt = await prisma.taskRunAttempt.findUnique({
where: {
id,
@@ -979,27 +1081,29 @@ class SharedQueueTasks {
return;
}
switch (attempt.status) {
case "CANCELED":
case "EXECUTING": {
logger.error("Invalid attempt status for execution payload retrieval", {
attemptId: id,
status: attempt.status,
});
return;
if (!skipStatusChecks) {
switch (attempt.status) {
case "CANCELED":
case "EXECUTING": {
logger.error("Invalid attempt status for execution payload retrieval", {
attemptId: id,
status: attempt.status,
});
return;
}
}
}
switch (attempt.taskRun.status) {
case "CANCELED":
case "EXECUTING":
case "INTERRUPTED": {
logger.error("Invalid run status for execution payload retrieval", {
attemptId: id,
runId: attempt.taskRunId,
status: attempt.taskRun.status,
});
return;
switch (attempt.taskRun.status) {
case "CANCELED":
case "EXECUTING":
case "INTERRUPTED": {
logger.error("Invalid run status for execution payload retrieval", {
attemptId: id,
runId: attempt.taskRunId,
status: attempt.taskRun.status,
});
return;
}
}
}
@@ -1150,7 +1254,11 @@ class SharedQueueTasks {
return;
}
return this.getExecutionPayloadFromAttempt(latestAttempt.id, setToExecuting, isRetrying);
return this.getExecutionPayloadFromAttempt({
id: latestAttempt.id,
setToExecuting,
isRetrying,
});
}
async getLazyAttemptPayload(
@@ -1225,13 +1333,13 @@ class SharedQueueTasks {
return;
}
await marqs?.heartbeatMessage(taskRunAttempt.taskRunId);
await this.#heartbeat(taskRunAttempt.taskRunId);
}
async taskRunHeartbeat(runId: string) {
logger.debug("[SharedQueueConsumer] taskRunHeartbeat()", { runId });
await marqs?.heartbeatMessage(runId);
await this.#heartbeat(runId);
}
public async taskRunFailed(completion: TaskRunFailedExecutionResult) {
@@ -1242,6 +1350,66 @@ class SharedQueueTasks {
await service.call(completion.id, completion);
}
async #heartbeat(runId: string) {
await marqs?.heartbeatMessage(runId);
try {
// There can be a lot of calls per minute and the data doesn't have to be accurate, so use the read replica
const taskRun = await $replica.taskRun.findFirst({
where: {
id: runId,
},
select: {
id: true,
status: true,
runtimeEnvironment: {
select: {
type: true,
},
},
lockedToVersion: {
select: {
supportsLazyAttempts: true,
},
},
},
});
if (!taskRun) {
logger.error("SharedQueueTasks.#heartbeat: Task run not found", {
runId,
});
return;
}
if (taskRun.runtimeEnvironment.type === "DEVELOPMENT") {
return;
}
if (isFinalRunStatus(taskRun.status)) {
logger.debug("SharedQueueTasks.#heartbeat: Task run is in final status", {
runId,
status: taskRun.status,
});
// Signal to exit any leftover containers
socketIo.coordinatorNamespace.emit("REQUEST_RUN_CANCELLATION", {
version: "v1",
runId: taskRun.id,
// Give the run a few seconds to exit to complete any flushing etc
delayInMs: taskRun.lockedToVersion?.supportsLazyAttempts ? 5_000 : undefined,
});
return;
}
} catch (error) {
logger.error("SharedQueueTasks.#heartbeat: Error signaling run cancellation", {
runId,
error: error instanceof Error ? error.message : error,
});
}
}
async #buildEnvironmentVariables(
environment: RuntimeEnvironment,
runId: string,
apps/webapp/app/v3/requeueTaskRun.server.ts
+38 -1
View File
@@ -6,11 +6,29 @@ import { FailedTaskRunService } from "./failedTaskRun.server";
import { BaseService } from "./services/baseService.server";
import { PrismaClientOrTransaction } from "~/db.server";
import { workerQueue } from "~/services/worker.server";
import { socketIo } from "./handleSocketIo.server";
export class RequeueTaskRunService extends BaseService {
public async call(runId: string) {
const taskRun = await this._prisma.taskRun.findUnique({
where: { id: runId },
where: {
id: runId,
},
select: {
id: true,
friendlyId: true,
status: true,
runtimeEnvironment: {
select: {
type: true,
},
},
lockedToVersion: {
select: {
supportsLazyAttempts: true,
},
},
},
});
if (!taskRun) {
@@ -76,6 +94,25 @@ export class RequeueTaskRunService extends BaseService {
await marqs?.acknowledgeMessage(taskRun.id);
try {
if (taskRun.runtimeEnvironment.type === "DEVELOPMENT") {
return;
}
// Signal to exit any leftover containers
socketIo.coordinatorNamespace.emit("REQUEST_RUN_CANCELLATION", {
version: "v1",
runId: taskRun.id,
// Give the run a few seconds to exit to complete any flushing etc
delayInMs: taskRun.lockedToVersion?.supportsLazyAttempts ? 5_000 : undefined,
});
} catch (error) {
logger.error("[RequeueTaskRunService] Error signaling run cancellation", {
runId: taskRun.id,
error: error instanceof Error ? error.message : error,
});
}
break;
}
default: {
apps/webapp/app/v3/services/restoreCheckpoint.server.ts
+20
View File
@@ -112,4 +112,24 @@ export class RestoreCheckpointService extends BaseService {
return checkpoint;
}
async getLastCheckpointEventIfUnrestored(runId: string) {
const event = await this._prisma.checkpointRestoreEvent.findFirst({
where: {
runId,
},
take: 1,
orderBy: {
createdAt: "desc",
},
});
if (!event) {
return;
}
if (event.type === "CHECKPOINT") {
return event;
}
}
}
apps/webapp/app/v3/services/resumeAttempt.server.ts
+31 -33
View File
@@ -2,23 +2,24 @@ import {
CoordinatorToPlatformMessages,
TaskRunExecution,
TaskRunExecutionResult,
WaitReason,
} from "@trigger.dev/core/v3";
import type { InferSocketMessageSchema } from "@trigger.dev/core/v3/zodSocket";
import { $transaction, PrismaClientOrTransaction } from "~/db.server";
import { logger } from "~/services/logger.server";
import { marqs } from "~/v3/marqs/index.server";
import { socketIo } from "../handleSocketIo.server";
import { SharedQueueMessageBody, sharedQueueTasks } from "../marqs/sharedQueueConsumer.server";
import { sharedQueueTasks } from "../marqs/sharedQueueConsumer.server";
import { BaseService } from "./baseService.server";
import { TaskRunAttempt } from "@trigger.dev/database";
import { isFinalRunStatus } from "../taskStatus";
export class ResumeAttemptService extends BaseService {
private _logger = logger;
public async call(
params: InferSocketMessageSchema<typeof CoordinatorToPlatformMessages, "READY_FOR_RESUME">
): Promise<void> {
logger.debug(`ResumeAttemptService.call()`, params);
this._logger.debug(`ResumeAttemptService.call()`, params);
await $transaction(this._prisma, async (tx) => {
const attempt = await tx.taskRunAttempt.findUnique({
@@ -77,16 +78,18 @@ export class ResumeAttemptService extends BaseService {
});
if (!attempt) {
logger.error("Could not find attempt", { attemptFriendlyId: params.attemptFriendlyId });
this._logger.error("Could not find attempt", params);
return;
}
this._logger = logger.child({
attemptId: attempt.id,
attemptFriendlyId: attempt.friendlyId,
taskRun: attempt.taskRun,
});
if (isFinalRunStatus(attempt.taskRun.status)) {
logger.error("Run is not resumable", {
attemptId: attempt.id,
runId: attempt.taskRunId,
status: attempt.taskRun.status,
});
this._logger.error("Run is not resumable");
return;
}
@@ -94,10 +97,7 @@ export class ResumeAttemptService extends BaseService {
switch (params.type) {
case "WAIT_FOR_DURATION": {
logger.debug("Sending duration wait resume message", {
attemptId: attempt.id,
attemptFriendlyId: params.attemptFriendlyId,
});
this._logger.debug("Sending duration wait resume message");
await this.#setPostResumeStatuses(attempt, tx);
@@ -114,13 +114,13 @@ export class ResumeAttemptService extends BaseService {
const dependentAttempt = attempt.dependencies[0].taskRun.attempts[0];
if (!dependentAttempt) {
logger.error("No dependent attempt", { attemptId: attempt.id });
this._logger.error("No dependent attempt");
return;
}
completedAttemptIds = [dependentAttempt.id];
} else {
logger.error("No task dependency", { attemptId: attempt.id });
this._logger.error("No task dependency");
return;
}
@@ -134,13 +134,13 @@ export class ResumeAttemptService extends BaseService {
const dependentBatchItems = attempt.batchDependencies[0].items;
if (!dependentBatchItems) {
logger.error("No dependent batch items", { attemptId: attempt.id });
this._logger.error("No dependent batch items");
return;
}
completedAttemptIds = dependentBatchItems.map((item) => item.taskRun.attempts[0]?.id);
} else {
logger.error("No batch dependency", { attemptId: attempt.id });
this._logger.error("No batch dependency");
return;
}
@@ -161,7 +161,7 @@ export class ResumeAttemptService extends BaseService {
tx: PrismaClientOrTransaction
) {
if (completedAttemptIds.length === 0) {
logger.error("No completed attempt IDs", { attemptId: attempt.id });
this._logger.error("No completed attempt IDs");
return;
}
@@ -184,38 +184,36 @@ export class ResumeAttemptService extends BaseService {
});
if (!completedAttempt) {
logger.error("Completed attempt not found", {
attemptId: attempt.id,
completedAttemptId,
});
this._logger.error("Completed attempt not found", { completedAttemptId });
await marqs?.acknowledgeMessage(attempt.taskRunId);
return;
}
const logger = this._logger.child({
completedAttemptId: completedAttempt.id,
completedAttemptFriendlyId: completedAttempt.friendlyId,
completedRunId: completedAttempt.taskRunId,
});
const completion = await sharedQueueTasks.getCompletionPayloadFromAttempt(
completedAttempt.id
);
if (!completion) {
logger.error("Failed to get completion payload", {
attemptId: attempt.id,
completedAttemptId,
});
logger.error("Failed to get completion payload");
await marqs?.acknowledgeMessage(attempt.taskRunId);
return;
}
completions.push(completion);
const executionPayload = await sharedQueueTasks.getExecutionPayloadFromAttempt(
completedAttempt.id
);
const executionPayload = await sharedQueueTasks.getExecutionPayloadFromAttempt({
id: completedAttempt.id,
skipStatusChecks: true, // already checked when getting the completion
});
if (!executionPayload) {
logger.error("Failed to get execution payload", {
attemptId: attempt.id,
completedAttemptId,
});
logger.error("Failed to get execution payload");
await marqs?.acknowledgeMessage(attempt.taskRunId);
return;
}
packages/core/src/v3/errors.ts
+61 -11
View File
@@ -312,8 +312,23 @@ export class GracefulExitTimeoutError extends Error {
}
}
type ErrorLink = {
name: string;
href: string;
// This allows us to easily add more complex logic on the frontend, e.g. display a button to open a contact form modal
magic?: "CONTACT_FORM";
};
type EnhanceError<T extends TaskRunError | ExceptionEventProperties> = T & { link?: ErrorLink };
const prettyInternalErrors: Partial<
Record<TaskRunInternalError["code"], { message: string; link?: { name: string; href: string } }>
Record<
TaskRunInternalError["code"],
{
message: string;
link?: ErrorLink;
}
>
> = {
TASK_PROCESS_OOM_KILLED: {
message:
@@ -331,10 +346,15 @@ const prettyInternalErrors: Partial<
href: links.docs.machines.home,
},
},
};
type EnhanceError<T extends TaskRunError | ExceptionEventProperties> = T & {
link?: { name: string; href: string };
TASK_PROCESS_SIGTERM: {
message:
"Your task exited after receiving SIGTERM but we don't know why. If this keeps happening, please get in touch so we can investigate.",
link: {
name: "Contact us",
href: links.site.contact,
magic: "CONTACT_FORM",
},
},
};
export function taskRunErrorEnhancer(error: TaskRunError): EnhanceError<TaskRunError> {
@@ -342,6 +362,14 @@ export function taskRunErrorEnhancer(error: TaskRunError): EnhanceError<TaskRunE
case "BUILT_IN_ERROR": {
if (error.name === "UnexpectedExitError") {
if (error.message.startsWith("Unexpected exit with code -1")) {
if (error.message.includes("SIGTERM")) {
return {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.TASK_PROCESS_SIGTERM,
...prettyInternalErrors.TASK_PROCESS_SIGTERM,
};
}
return {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.TASK_PROCESS_MAYBE_OOM_KILLED,
@@ -359,6 +387,14 @@ export function taskRunErrorEnhancer(error: TaskRunError): EnhanceError<TaskRunE
}
case "INTERNAL_ERROR": {
if (error.code === TaskRunErrorCodes.TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE) {
if (error.message?.includes("SIGTERM")) {
return {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.TASK_PROCESS_SIGTERM,
...prettyInternalErrors.TASK_PROCESS_SIGTERM,
};
}
return {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.TASK_PROCESS_MAYBE_OOM_KILLED,
@@ -396,7 +432,8 @@ export function exceptionEventEnhancer(
break;
}
case TaskRunErrorCodes.TASK_PROCESS_MAYBE_OOM_KILLED:
case TaskRunErrorCodes.TASK_PROCESS_OOM_KILLED: {
case TaskRunErrorCodes.TASK_PROCESS_OOM_KILLED:
case TaskRunErrorCodes.TASK_PROCESS_SIGTERM: {
return {
...exception,
...prettyInternalErrors[exception.type],
@@ -411,16 +448,23 @@ export function internalErrorFromUnexpectedExit(
error: UnexpectedExitError,
dockerMode = true
): TaskRunInternalError {
const internalError = {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE,
message: `Process exited with code ${error.code} after signal ${error.signal}.`,
stackTrace: error.stderr,
} satisfies TaskRunInternalError;
if (error.code === 137) {
if (dockerMode) {
return {
type: "INTERNAL_ERROR",
...internalError,
code: TaskRunErrorCodes.TASK_PROCESS_OOM_KILLED,
};
} else {
// Note: containerState reason and message could be checked to clarify the error, maybe the task monitor should be allowed to override these
return {
type: "INTERNAL_ERROR",
...internalError,
code: TaskRunErrorCodes.TASK_PROCESS_MAYBE_OOM_KILLED,
};
}
@@ -428,15 +472,21 @@ export function internalErrorFromUnexpectedExit(
if (error.stderr?.includes("OOMErrorHandler")) {
return {
type: "INTERNAL_ERROR",
...internalError,
code: TaskRunErrorCodes.TASK_PROCESS_OOM_KILLED,
};
}
if (error.signal === "SIGTERM") {
return {
...internalError,
code: TaskRunErrorCodes.TASK_PROCESS_SIGTERM,
};
}
return {
type: "INTERNAL_ERROR",
...internalError,
code: TaskRunErrorCodes.TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE,
message: `Process exited with code ${error.code} after signal ${error.signal}.`,
};
}
packages/core/src/v3/schemas/common.ts
+3 -23
View File
@@ -77,29 +77,6 @@ export const TaskRunStringError = z.object({
export type TaskRunStringError = z.infer<typeof TaskRunStringError>;
export const TaskRunErrorCodes = {
COULD_NOT_FIND_EXECUTOR: "COULD_NOT_FIND_EXECUTOR",
COULD_NOT_FIND_TASK: "COULD_NOT_FIND_TASK",
COULD_NOT_IMPORT_TASK: "COULD_NOT_IMPORT_TASK",
CONFIGURED_INCORRECTLY: "CONFIGURED_INCORRECTLY",
TASK_ALREADY_RUNNING: "TASK_ALREADY_RUNNING",
TASK_EXECUTION_FAILED: "TASK_EXECUTION_FAILED",
TASK_EXECUTION_ABORTED: "TASK_EXECUTION_ABORTED",
TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE: "TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE",
TASK_PROCESS_SIGKILL_TIMEOUT: "TASK_PROCESS_SIGKILL_TIMEOUT",
TASK_PROCESS_OOM_KILLED: "TASK_PROCESS_OOM_KILLED",
TASK_PROCESS_MAYBE_OOM_KILLED: "TASK_PROCESS_MAYBE_OOM_KILLED",
TASK_RUN_CANCELLED: "TASK_RUN_CANCELLED",
TASK_OUTPUT_ERROR: "TASK_OUTPUT_ERROR",
HANDLE_ERROR_ERROR: "HANDLE_ERROR_ERROR",
GRACEFUL_EXIT_TIMEOUT: "GRACEFUL_EXIT_TIMEOUT",
TASK_RUN_CRASHED: "TASK_RUN_CRASHED",
MAX_DURATION_EXCEEDED: "MAX_DURATION_EXCEEDED",
DISK_SPACE_EXCEEDED: "DISK_SPACE_EXCEEDED",
POD_EVICTED: "POD_EVICTED",
POD_UNKNOWN_ERROR: "POD_UNKNOWN_ERROR",
} as const;
export const TaskRunInternalError = z.object({
type: z.literal("INTERNAL_ERROR"),
code: z.enum([
@@ -112,6 +89,7 @@ export const TaskRunInternalError = z.object({
"TASK_EXECUTION_ABORTED",
"TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE",
"TASK_PROCESS_SIGKILL_TIMEOUT",
"TASK_PROCESS_SIGTERM",
"TASK_PROCESS_OOM_KILLED",
"TASK_PROCESS_MAYBE_OOM_KILLED",
"TASK_RUN_CANCELLED",
@@ -131,6 +109,8 @@ export const TaskRunInternalError = z.object({
export type TaskRunInternalError = z.infer<typeof TaskRunInternalError>;
export const TaskRunErrorCodes = TaskRunInternalError.shape.code.enum;
export const TaskRunError = z.discriminatedUnion("type", [
TaskRunBuiltInError,
TaskRunCustomErrorObject,