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

v3: fix unfreezable state crashes for runs with multiple waits (#1253)

Browse Source 
* support named capture groups

* write crash errors to attempt.error

* make restored pod names unique per checkpoint

* use last eight characters of checkpoint id instead

* add more chaos monkey env vars

* Ignore unfreezable states

* prevent excessive queue config parsing errors

* handle dependency resume edge case

* better entry point logging

* ignore checkpoint cancellation timeouts

* add missing idempotency keys to wait for dep replays

* remove checkpoints between attempts

* fix retry container names on kubernetes

* add changeset

* fix types

* bring back internal duration timers
This commit is contained in:
nicktrn
2024-08-13 11:44:40 +01:00
committed by GitHub
parent da6ce3c8d5
commit 5cf90da722
22 changed files with 461 additions and 350 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/hungry-sloths-promise.md
+12
View File
@@ -0,0 +1,12 @@
---
"@trigger.dev/core-apps": patch
"trigger.dev": patch
"@trigger.dev/core": patch
---
Fix issues that could result in unreezable state run crashes. Details:
- Never checkpoint between attempts
- Some messages and socket data now include attempt numbers
- Remove attempt completion replays
- Additional prod entry point logging
- Fail runs that receive deprecated (pre-lazy attempt) execute messages
apps/coordinator/src/chaosMonkey.ts
+7 -3
View File
@@ -12,7 +12,11 @@ export class ChaosMonkey {
private chaosEventRate = 0.2;
private delayInSeconds = 45;
constructor(private enabled = false) {
constructor(
private enabled = false,
private disableErrors = false,
private disableDelays = false
) {
if (this.enabled) {
console.log("🍌 Chaos monkey enabled");
}
@@ -32,8 +36,8 @@ export class ChaosMonkey {
async call({
$,
throwErrors = true,
addDelays = true,
throwErrors = !this.disableErrors,
addDelays = !this.disableDelays,
}: {
$?: Execa$<string>;
throwErrors?: boolean;
apps/coordinator/src/checkpointer.ts
+8 -3
View File
@@ -17,6 +17,7 @@ type CheckpointAndPushOptions = {
projectRef: string;
deploymentVersion: string;
shouldHeartbeat?: boolean;
attemptNumber?: number;
};
type CheckpointAndPushResult =
@@ -258,6 +259,7 @@ export class Checkpointer {
leaveRunning = true, // This mirrors kubernetes behaviour more accurately
projectRef,
deploymentVersion,
attemptNumber,
}: CheckpointAndPushOptions): Promise<CheckpointAndPushResult> {
this.#logger.log("Checkpointing with backoff", {
runId,
@@ -297,6 +299,7 @@ export class Checkpointer {
leaveRunning,
projectRef,
deploymentVersion,
attemptNumber,
});
if (result.success) {
@@ -359,6 +362,7 @@ export class Checkpointer {
leaveRunning = true, // This mirrors kubernetes behaviour more accurately
projectRef,
deploymentVersion,
attemptNumber,
}: CheckpointAndPushOptions): Promise<CheckpointAndPushResult> {
await this.init();
@@ -367,6 +371,7 @@ export class Checkpointer {
leaveRunning,
projectRef,
deploymentVersion,
attemptNumber,
};
if (!this.#dockerMode && !this.#canCheckpoint) {
@@ -417,7 +422,7 @@ export class Checkpointer {
this.#logger.log("Checkpointing:", { options });
const containterName = this.#getRunContainerName(runId);
const containterName = this.#getRunContainerName(runId, attemptNumber);
// Create checkpoint (docker)
if (this.#dockerMode) {
@@ -581,7 +586,7 @@ export class Checkpointer {
return this.#failedCheckpoints.has(runId);
}
#getRunContainerName(suffix: string) {
return `task-run-${suffix}`;
#getRunContainerName(suffix: string, attemptNumber?: number) {
return `task-run-${suffix}${attemptNumber && attemptNumber > 1 ? `-att${attemptNumber}` : ""}`;
}
}
apps/coordinator/src/index.ts
+39 -4
View File
@@ -49,7 +49,11 @@ const PLATFORM_SECRET = process.env.PLATFORM_SECRET || "coordinator-secret";
const SECURE_CONNECTION = ["1", "true"].includes(process.env.SECURE_CONNECTION ?? "false");
const logger = new SimpleLogger(`[${NODE_NAME}]`);
const chaosMonkey = new ChaosMonkey(!!process.env.CHAOS_MONKEY_ENABLED);
const chaosMonkey = new ChaosMonkey(
!!process.env.CHAOS_MONKEY_ENABLED,
!!process.env.CHAOS_MONKEY_DISABLE_ERRORS,
!!process.env.CHAOS_MONKEY_DISABLE_DELAYS
);
class TaskCoordinator {
#httpServer: ReturnType<typeof createServer>;
@@ -290,6 +294,7 @@ class TaskCoordinator {
setSocketDataFromHeader("projectRef", "x-trigger-project-ref");
setSocketDataFromHeader("runId", "x-trigger-run-id");
setSocketDataFromHeader("attemptFriendlyId", "x-trigger-attempt-friendly-id", false);
setSocketDataFromHeader("attemptNumber", "x-trigger-attempt-number", false);
setSocketDataFromHeader("envId", "x-trigger-env-id");
setSocketDataFromHeader("deploymentId", "x-trigger-deployment-id");
setSocketDataFromHeader("deploymentVersion", "x-trigger-deployment-version");
@@ -306,6 +311,10 @@ class TaskCoordinator {
onConnection: async (socket, handler, sender) => {
const logger = new SimpleLogger(`[prod-worker][${socket.id}]`);
const getAttemptNumber = () => {
return socket.data.attemptNumber ? parseInt(socket.data.attemptNumber) : undefined;
};
const crashRun = async (error: { name: string; message: string; stack?: string }) => {
try {
this.#platformSocket?.send("RUN_CRASHED", {
@@ -381,6 +390,10 @@ class TaskCoordinator {
socket.data.attemptFriendlyId = attemptFriendlyId;
};
const updateAttemptNumber = (attemptNumber: string | number) => {
socket.data.attemptNumber = String(attemptNumber);
};
this.#platformSocket?.send("LOG", {
metadata: socket.data,
text: "connected",
@@ -430,6 +443,7 @@ class TaskCoordinator {
});
updateAttemptFriendlyId(executionAck.payload.execution.attempt.id);
updateAttemptNumber(executionAck.payload.execution.attempt.number);
} catch (error) {
logger.error("Error", { error });
@@ -505,11 +519,17 @@ class TaskCoordinator {
updateAttemptFriendlyId(message.attemptFriendlyId);
this.#platformSocket?.send("READY_FOR_RESUME", message);
if (message.version === "v2") {
updateAttemptNumber(message.attemptNumber);
}
this.#platformSocket?.send("READY_FOR_RESUME", { ...message, version: "v1" });
});
// MARK: RUN COMPLETED
socket.on("TASK_RUN_COMPLETED", async ({ completion, execution }, callback) => {
socket.on("TASK_RUN_COMPLETED", async (message, callback) => {
const { completion, execution } = message;
logger.log("completed task", { completionId: completion.id });
// Cancel all in-progress checkpoints (if any)
@@ -518,8 +538,10 @@ class TaskCoordinator {
await chaosMonkey.call({ throwErrors: false });
const completeWithoutCheckpoint = (shouldExit: boolean) => {
const supportsRetryCheckpoints = message.version === "v1";
this.#platformSocket?.send("TASK_RUN_COMPLETED", {
version: "v1",
version: supportsRetryCheckpoints ? "v1" : "v2",
execution,
completion,
});
@@ -549,6 +571,11 @@ class TaskCoordinator {
return;
}
if (message.version === "v2") {
completeWithoutCheckpoint(true);
return;
}
const { canCheckpoint, willSimulate } = await this.#checkpointer.init();
const willCheckpointAndRestore = canCheckpoint || willSimulate;
@@ -681,6 +708,7 @@ class TaskCoordinator {
runId: socket.data.runId,
projectRef: socket.data.projectRef,
deploymentVersion: socket.data.deploymentVersion,
attemptNumber: getAttemptNumber(),
});
if (!checkpoint) {
@@ -752,6 +780,7 @@ class TaskCoordinator {
runId: socket.data.runId,
projectRef: socket.data.projectRef,
deploymentVersion: socket.data.deploymentVersion,
attemptNumber: getAttemptNumber(),
});
if (!checkpoint) {
@@ -821,6 +850,7 @@ class TaskCoordinator {
runId: socket.data.runId,
projectRef: socket.data.projectRef,
deploymentVersion: socket.data.deploymentVersion,
attemptNumber: getAttemptNumber(),
});
if (!checkpoint) {
@@ -905,6 +935,7 @@ class TaskCoordinator {
}
updateAttemptFriendlyId(createAttempt.executionPayload.execution.attempt.id);
updateAttemptNumber(createAttempt.executionPayload.execution.attempt.number);
callback({
success: true,
@@ -924,6 +955,10 @@ class TaskCoordinator {
if (message.attemptFriendlyId) {
updateAttemptFriendlyId(message.attemptFriendlyId);
}
if (message.attemptNumber) {
updateAttemptNumber(message.attemptNumber);
}
});
},
onDisconnect: async (socket, handler, sender, logger) => {
apps/docker-provider/src/index.ts
+4 -4
View File
@@ -109,7 +109,7 @@ class DockerTaskOperations implements TaskOperations {
async create(opts: TaskOperationsCreateOptions) {
await this.init();
const containerName = this.#getRunContainerName(opts.runId);
const containerName = this.#getRunContainerName(opts.runId, opts.nextAttemptNumber);
const runArgs = [
"run",
@@ -150,7 +150,7 @@ class DockerTaskOperations implements TaskOperations {
async restore(opts: TaskOperationsRestoreOptions) {
await this.init();
const containerName = this.#getRunContainerName(opts.runId);
const containerName = this.#getRunContainerName(opts.runId, opts.attemptNumber);
if (!this.#canCheckpoint || this.opts.forceSimulate) {
logger.log("Simulating restore");
@@ -195,8 +195,8 @@ class DockerTaskOperations implements TaskOperations {
return `task-index-${suffix}`;
}
#getRunContainerName(suffix: string) {
return `task-run-${suffix}`;
#getRunContainerName(suffix: string, attemptNumber?: number) {
return `task-run-${suffix}${attemptNumber && attemptNumber > 1 ? `-att${attemptNumber}` : ""}`;
}
async #sendPostStart(containerName: string): Promise<void> {
apps/docker-provider/tsconfig.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"compilerOptions": {
"target": "es2016",
"target": "es2018",
"module": "commonjs",
"esModuleInterop": true,
"forceConsistentCasingInFileNames": true,
apps/kubernetes-provider/src/index.ts
+7 -5
View File
@@ -139,10 +139,12 @@ class KubernetesTaskOperations implements TaskOperations {
}
async create(opts: TaskOperationsCreateOptions) {
const containerName = this.#getRunContainerName(opts.runId, opts.nextAttemptNumber);
await this.#createPod(
{
metadata: {
name: this.#getRunContainerName(opts.runId),
name: containerName,
namespace: this.#namespace.metadata.name,
labels: {
...this.#getSharedLabels(opts),
@@ -157,7 +159,7 @@ class KubernetesTaskOperations implements TaskOperations {
terminationGracePeriodSeconds: 60 * 60,
containers: [
{
name: this.#getRunContainerName(opts.runId),
name: containerName,
image: opts.image,
ports: [
{
@@ -211,7 +213,7 @@ class KubernetesTaskOperations implements TaskOperations {
await this.#createPod(
{
metadata: {
name: `${this.#getRunContainerName(opts.runId)}-${randomUUID().slice(0, 8)}`,
name: `${this.#getRunContainerName(opts.runId)}-${opts.checkpointId.slice(-8)}`,
namespace: this.#namespace.metadata.name,
labels: {
...this.#getSharedLabels(opts),
@@ -514,8 +516,8 @@ class KubernetesTaskOperations implements TaskOperations {
return `task-index-${suffix}`;
}
#getRunContainerName(suffix: string) {
return `task-run-${suffix}`;
#getRunContainerName(suffix: string, attemptNumber?: number) {
return `task-run-${suffix}${attemptNumber && attemptNumber > 1 ? `-att${attemptNumber}` : ""}`;
}
#getPrePullContainerName(suffix: string) {
apps/webapp/app/v3/failedTaskRun.server.ts
+2 -4
View File
@@ -1,11 +1,9 @@
import { TaskRunFailedExecutionResult } from "@trigger.dev/core/v3";
import { TaskRunStatus } from "@trigger.dev/database";
import { logger } from "~/services/logger.server";
import { createExceptionPropertiesFromError, eventRepository } from "./eventRepository.server";
import { BaseService } from "./services/baseService.server";
import { FinalizeTaskRunService } from "./services/finalizeTaskRun.server";
const FAILABLE_TASK_RUN_STATUSES: TaskRunStatus[] = ["EXECUTING", "PENDING", "WAITING_FOR_DEPLOY"];
import { FAILABLE_RUN_STATUSES } from "./taskStatus";
export class FailedTaskRunService extends BaseService {
public async call(anyRunId: string, completion: TaskRunFailedExecutionResult) {
@@ -27,7 +25,7 @@ export class FailedTaskRunService extends BaseService {
return;
}
if (!FAILABLE_TASK_RUN_STATUSES.includes(taskRun.status)) {
if (!FAILABLE_RUN_STATUSES.includes(taskRun.status)) {
logger.error("[FailedTaskRunService] Task run is not in a failable state", {
taskRun,
completion,
apps/webapp/app/v3/handleSocketIo.server.ts
+1
View File
@@ -128,6 +128,7 @@ function createCoordinatorNamespace(io: Server) {
completion: message.completion,
execution: message.execution,
checkpoint: message.checkpoint,
supportsRetryCheckpoints: message.version === "v1",
});
},
TASK_RUN_FAILED_TO_RUN: async (message) => {
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
+12 -8
View File
@@ -54,6 +54,7 @@ export const SharedQueueMessageBody = z.discriminatedUnion("type", [
type: z.literal("EXECUTE"),
taskIdentifier: z.string(),
checkpointEventId: z.string().optional(),
retryCheckpointsDisabled: z.boolean().optional(),
}),
WithTraceContext.extend({
type: z.literal("RESUME"),
@@ -294,12 +295,9 @@ export class SharedQueueConsumer {
const retryingFromCheckpoint = !!messageBody.data.checkpointEventId;
const EXECUTABLE_RUN_STATUSES: {
fromCheckpoint: TaskRunStatus[];
withoutCheckpoint: TaskRunStatus[];
} = {
fromCheckpoint: ["WAITING_TO_RESUME"],
withoutCheckpoint: ["PENDING", "RETRYING_AFTER_FAILURE"],
const EXECUTABLE_RUN_STATUSES = {
fromCheckpoint: ["WAITING_TO_RESUME"] satisfies TaskRunStatus[],
withoutCheckpoint: ["PENDING", "RETRYING_AFTER_FAILURE"] satisfies TaskRunStatus[],
};
if (
@@ -474,7 +472,10 @@ export class SharedQueueConsumer {
? lockedTaskRun.attempts[0].number + 1
: 1;
const isRetry = lockedTaskRun.status === "WAITING_TO_RESUME" && nextAttemptNumber > 1;
const isRetry =
nextAttemptNumber > 1 &&
(lockedTaskRun.status === "WAITING_TO_RESUME" ||
lockedTaskRun.status === "RETRYING_AFTER_FAILURE");
try {
if (messageBody.data.checkpointEventId) {
@@ -515,11 +516,13 @@ export class SharedQueueConsumer {
}
}
if (isRetry) {
if (isRetry && !messageBody.data.retryCheckpointsDisabled) {
socketIo.coordinatorNamespace.emit("READY_FOR_RETRY", {
version: "v1",
runId: lockedTaskRun.id,
});
// Retries for workers with disabled retry checkpoints will be handled just like normal attempts
} else {
const machineConfig = lockedTaskRun.lockedBy?.machineConfig;
const machine = machinePresetFromConfig(machineConfig ?? {});
@@ -531,6 +534,7 @@ export class SharedQueueConsumer {
image: deployment.imageReference,
version: deployment.version,
machine,
nextAttemptNumber,
// identifiers
id: "placeholder", // TODO: Remove this completely in a future release
envId: lockedTaskRun.runtimeEnvironment.id,
apps/webapp/app/v3/services/completeAttempt.server.ts
+59 -30
View File
@@ -38,11 +38,13 @@ export class CompleteAttemptService extends BaseService {
execution,
env,
checkpoint,
supportsRetryCheckpoints,
}: {
completion: TaskRunExecutionResult;
execution: TaskRunExecution;
env?: AuthenticatedEnvironment;
checkpoint?: CheckpointData;
supportsRetryCheckpoints?: boolean;
}): Promise<"COMPLETED" | "RETRIED"> {
const taskRunAttempt = await findAttempt(this._prisma, execution.attempt.id);
@@ -95,13 +97,14 @@ export class CompleteAttemptService extends BaseService {
if (completion.ok) {
return await this.#completeAttemptSuccessfully(completion, taskRunAttempt, env);
} else {
return await this.#completeAttemptFailed(
return await this.#completeAttemptFailed({
completion,
execution,
taskRunAttempt,
env,
checkpoint
);
checkpoint,
supportsRetryCheckpoints,
});
}
}
@@ -152,13 +155,21 @@ export class CompleteAttemptService extends BaseService {
return "COMPLETED";
}
async #completeAttemptFailed(
completion: TaskRunFailedExecutionResult,
execution: TaskRunExecution,
taskRunAttempt: NonNullable<FoundAttempt>,
env?: AuthenticatedEnvironment,
checkpoint?: CheckpointData
): Promise<"COMPLETED" | "RETRIED"> {
async #completeAttemptFailed({
completion,
execution,
taskRunAttempt,
env,
checkpoint,
supportsRetryCheckpoints,
}: {
completion: TaskRunFailedExecutionResult;
execution: TaskRunExecution;
taskRunAttempt: NonNullable<FoundAttempt>;
env?: AuthenticatedEnvironment;
checkpoint?: CheckpointData;
supportsRetryCheckpoints?: boolean;
}): Promise<"COMPLETED" | "RETRIED"> {
if (
completion.error.type === "INTERNAL_ERROR" &&
completion.error.code === "TASK_RUN_CANCELLED"
@@ -243,12 +254,13 @@ export class CompleteAttemptService extends BaseService {
}
if (!checkpoint) {
await this.#retryAttempt(
taskRunAttempt.taskRun,
completion.retry.timestamp,
undefined,
taskRunAttempt.backgroundWorker.supportsLazyAttempts
);
await this.#retryAttempt({
run: taskRunAttempt.taskRun,
retryTimestamp: completion.retry.timestamp,
supportsLazyAttempts: taskRunAttempt.backgroundWorker.supportsLazyAttempts,
supportsRetryCheckpoints,
});
return "RETRIED";
}
@@ -263,7 +275,7 @@ export class CompleteAttemptService extends BaseService {
},
});
if (!checkpointCreateResult) {
if (!checkpointCreateResult.success) {
logger.error("Failed to create checkpoint", { checkpoint, execution: execution.run.id });
const finalizeService = new FinalizeTaskRunService();
@@ -276,11 +288,13 @@ export class CompleteAttemptService extends BaseService {
return "COMPLETED";
}
await this.#retryAttempt(
taskRunAttempt.taskRun,
completion.retry.timestamp,
checkpointCreateResult.event.id
);
await this.#retryAttempt({
run: taskRunAttempt.taskRun,
retryTimestamp: completion.retry.timestamp,
checkpointEventId: checkpointCreateResult.event.id,
supportsLazyAttempts: taskRunAttempt.backgroundWorker.supportsLazyAttempts,
supportsRetryCheckpoints,
});
return "RETRIED";
} else {
@@ -352,13 +366,27 @@ export class CompleteAttemptService extends BaseService {
}
}
async #retryAttempt(
run: TaskRun,
retryTimestamp: number,
checkpointEventId?: string,
supportsLazyAttempts?: boolean
) {
if (checkpointEventId || !supportsLazyAttempts) {
async #retryAttempt({
run,
retryTimestamp,
checkpointEventId,
supportsLazyAttempts,
supportsRetryCheckpoints,
}: {
run: TaskRun;
retryTimestamp: number;
checkpointEventId?: string;
supportsLazyAttempts: boolean;
supportsRetryCheckpoints?: boolean;
}) {
if (checkpointEventId || !supportsLazyAttempts || !supportsRetryCheckpoints) {
if (!supportsRetryCheckpoints && checkpointEventId) {
logger.error("Worker does not support retry checkpoints, but a checkpoint was created", {
runId: run.id,
checkpointEventId,
});
}
// Workers without lazy attempt support always need to go through the queue, which is where the attempt is created
// We have to replace a potential RESUME with EXECUTE to correctly retry the attempt
return await marqs?.replaceMessage(
@@ -366,7 +394,8 @@ export class CompleteAttemptService extends BaseService {
{
type: "EXECUTE",
taskIdentifier: run.taskIdentifier,
checkpointEventId: checkpointEventId,
checkpointEventId: supportsRetryCheckpoints ? checkpointEventId : undefined,
retryCheckpointsDisabled: !supportsRetryCheckpoints,
},
retryTimestamp
);
apps/webapp/app/v3/services/createCheckpoint.server.ts
+23 -13
View File
@@ -6,7 +6,6 @@ import { generateFriendlyId } from "../friendlyIdentifiers";
import { marqs } from "~/v3/marqs/index.server";
import { CreateCheckpointRestoreEventService } from "./createCheckpointRestoreEvent.server";
import { BaseService } from "./baseService.server";
import { CrashTaskRunService } from "./crashTaskRun.server";
import { isFinalRunStatus, isFreezableAttemptStatus, isFreezableRunStatus } from "../taskStatus";
export class CreateCheckpointService extends BaseService {
@@ -17,11 +16,15 @@ export class CreateCheckpointService extends BaseService {
>
): Promise<
| {
success: true;
checkpoint: Checkpoint;
event: CheckpointRestoreEvent;
keepRunAlive: boolean;
}
| undefined
| {
success: false;
keepRunAlive?: boolean;
}
> {
logger.debug(`Creating checkpoint`, params);
@@ -46,7 +49,10 @@ export class CreateCheckpointService extends BaseService {
if (!attempt) {
logger.error("Attempt not found", { attemptFriendlyId: params.attemptFriendlyId });
return;
return {
success: false,
};
}
if (
@@ -64,14 +70,10 @@ export class CreateCheckpointService extends BaseService {
},
});
// This should only affect CLIs < beta.24, in very limited scenarios
const service = new CrashTaskRunService(this._prisma);
await service.call(attempt.taskRunId, {
crashAttempts: true,
reason: "Unfreezable state: Please upgrade your CLI",
});
return;
return {
success: false,
keepRunAlive: true,
};
}
const imageRef = attempt.backgroundWorker.deployment?.imageReference;
@@ -81,7 +83,10 @@ export class CreateCheckpointService extends BaseService {
attemptId: attempt.id,
workerId: attempt.backgroundWorker.id,
});
return;
return {
success: false,
};
}
const checkpoint = await this._prisma.checkpoint.create({
@@ -90,6 +95,7 @@ export class CreateCheckpointService extends BaseService {
runtimeEnvironmentId: attempt.taskRun.runtimeEnvironmentId,
projectId: attempt.taskRun.projectId,
attemptId: attempt.id,
attemptNumber: attempt.number,
runId: attempt.taskRunId,
location: params.location,
type: params.docker ? "DOCKER" : "KUBERNETES",
@@ -175,7 +181,10 @@ export class CreateCheckpointService extends BaseService {
checkpointId: checkpoint.id,
});
await marqs?.acknowledgeMessage(attempt.taskRunId);
return;
return {
success: false,
};
}
if (reason.type === "WAIT_FOR_DURATION") {
@@ -191,6 +200,7 @@ export class CreateCheckpointService extends BaseService {
}
return {
success: true,
checkpoint,
event: checkpointEvent,
keepRunAlive,
apps/webapp/app/v3/services/restoreCheckpoint.server.ts
+1
View File
@@ -100,6 +100,7 @@ export class RestoreCheckpointService extends BaseService {
reason: checkpoint.reason ?? undefined,
imageRef: checkpoint.imageRef,
machine,
attemptNumber: checkpoint.attemptNumber ?? undefined,
// identifiers
checkpointId: checkpoint.id,
envId: checkpoint.runtimeEnvironment.id,
apps/webapp/app/v3/services/resumeBatchRun.server.ts
+10 -10
View File
@@ -61,16 +61,7 @@ export class ResumeBatchRunService extends BaseService {
const dependentRun = batchRun.dependentTaskAttempt.taskRun;
if (batchRun.dependentTaskAttempt.status === "PAUSED") {
if (!batchRun.checkpointEventId) {
logger.error("Can't resume paused attempt without checkpoint event", {
batchRunId: batchRun.id,
});
await marqs?.acknowledgeMessage(dependentRun.id);
return;
}
if (batchRun.dependentTaskAttempt.status === "PAUSED" && batchRun.checkpointEventId) {
await marqs?.enqueueMessage(
environment,
dependentRun.queue,
@@ -87,6 +78,15 @@ export class ResumeBatchRunService extends BaseService {
dependentRun.concurrencyKey ?? undefined
);
} else {
if (batchRun.dependentTaskAttempt.status === "PAUSED" && !batchRun.checkpointEventId) {
// In case of race conditions and other bugs, the status can be PAUSED without a checkpoint event
// The worker may still be up, so we will try to resume the dependent attempt by sending a message to the worker (on dequeue)
logger.error("Batch run resume: Attempt is paused but there's no checkpoint event", {
batchRunId: batchRun.id,
dependentTaskAttemptId: batchRun.dependentTaskAttempt.id,
});
}
await marqs?.replaceMessage(dependentRun.id, {
type: "RESUME",
completedAttemptIds: batchRun.items.map((item) => item.taskRunAttemptId).filter(Boolean),
apps/webapp/app/v3/services/resumeTaskDependency.server.ts
+10 -10
View File
@@ -38,16 +38,7 @@ export class ResumeTaskDependencyService extends BaseService {
const dependentRun = dependency.dependentAttempt.taskRun;
if (dependency.dependentAttempt.status === "PAUSED") {
if (!dependency.checkpointEventId) {
logger.error("Can't resume paused attempt without checkpoint event", {
attemptId: dependency.id,
});
await marqs?.acknowledgeMessage(dependentRun.id);
return;
}
if (dependency.dependentAttempt.status === "PAUSED" && dependency.checkpointEventId) {
await marqs?.enqueueMessage(
dependency.taskRun.runtimeEnvironment,
dependentRun.queue,
@@ -64,6 +55,15 @@ export class ResumeTaskDependencyService extends BaseService {
dependentRun.concurrencyKey ?? undefined
);
} else {
if (dependency.dependentAttempt.status === "PAUSED" && !dependency.checkpointEventId) {
// In case of race conditions and other bugs, the status can be PAUSED without a checkpoint event
// The worker may still be up, so we will try to resume the dependent attempt by sending a message to the worker (on dequeue)
logger.warn("Task dependency resume: Attempt is paused but there's no checkpoint event", {
attemptId: dependency.id,
dependentAttemptId: dependency.dependentAttempt.id,
});
}
await marqs?.replaceMessage(dependentRun.id, {
type: "RESUME",
completedAttemptIds: [sourceTaskAttemptId],
apps/webapp/app/v3/services/triggerTask.server.ts
+1 -1
View File
@@ -445,7 +445,7 @@ export class TriggerTaskService extends BaseService {
return defaultQueueName;
}
const queueConfig = QueueOptions.optional().safeParse(task.queueConfig);
const queueConfig = QueueOptions.optional().nullable().safeParse(task.queueConfig);
if (!queueConfig.success) {
console.log("Failed to get queue name: Invalid queue config", {
apps/webapp/app/v3/taskStatus.ts
+7
View File
@@ -69,3 +69,10 @@ export function isRestorableRunStatus(status: TaskRunStatus): boolean {
export function isRestorableAttemptStatus(status: TaskRunAttemptStatus): boolean {
return RESTORABLE_ATTEMPT_STATUSES.includes(status);
}
export const FAILABLE_RUN_STATUSES = [
"EXECUTING",
"PENDING",
"WAITING_FOR_DEPLOY",
"RETRYING_AFTER_FAILURE",
] satisfies TaskRunStatus[];
packages/cli-v3/src/workers/prod/entry-point.ts
+229 -245
View File
@@ -27,6 +27,7 @@ import { randomUUID } from "node:crypto";
import { readFile } from "node:fs/promises";
import { createServer } from "node:http";
import { setTimeout as timeout } from "node:timers/promises";
import { Evt } from "evt";
declare const __PROJECT_CONFIG__: Config;
@@ -58,11 +59,14 @@ class ProdWorker {
private completed = new Set<string>();
private paused = false;
private attemptFriendlyId?: string;
private attemptNumber?: number;
private nextResumeAfter?: WaitReason;
private waitForPostStart = false;
private connectionCount = 0;
private restoreNotification = Evt.create();
private waitForTaskReplay:
| {
idempotencyKey: string;
@@ -82,16 +86,6 @@ class ProdWorker {
idempotencyKey: string;
}
| undefined;
private submitAttemptCompletionReplay:
| {
idempotencyKey: string;
message: {
execution: ProdTaskRunExecution;
completion: TaskRunExecutionResult;
};
attempt: number;
}
| undefined;
private durationResumeFallback:
| {
idempotencyKey: string;
@@ -190,128 +184,156 @@ class ProdWorker {
}
// MARK: TASK WAIT
async #waitForTaskHandler(message: OnWaitForTaskMessage, replayIdempotencyKey?: string) {
const waitForTask = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Wait for task with backoff", { retry });
#waitForTaskHandlerFactory(workerId?: string) {
return async (message: OnWaitForTaskMessage, replayIdempotencyKey?: string) => {
logger.log("onWaitForTask", { workerId, message });
if (!this.attemptFriendlyId) {
logger.error("Failed to send wait message, attempt friendly ID not set", { message });
if (this.nextResumeAfter) {
logger.error("Already waiting for resume, skipping wait for task", {
nextResumeAfter: this.nextResumeAfter,
});
throw new ExponentialBackoff.StopRetrying("No attempt ID");
return;
}
return await this.#coordinatorSocket.socket.timeout(20_000).emitWithAck("WAIT_FOR_TASK", {
version: "v2",
friendlyId: message.friendlyId,
attemptFriendlyId: this.attemptFriendlyId,
});
});
const waitForTask = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Wait for task with backoff", { retry });
if (!waitForTask.success) {
logger.error("Failed to wait for task with backoff", {
cause: waitForTask.cause,
error: waitForTask.error,
});
if (!this.attemptFriendlyId) {
logger.error("Failed to send wait message, attempt friendly ID not set", { message });
this.#emitUnrecoverableError(
"WaitForTaskFailed",
`${waitForTask.cause}: ${waitForTask.error}`
);
return;
}
const { willCheckpointAndRestore } = waitForTask.result;
await this.#prepareForWait("WAIT_FOR_TASK", willCheckpointAndRestore);
if (willCheckpointAndRestore) {
// We need to replay this on next connection if we don't receive RESUME_AFTER_DEPENDENCY within a reasonable time
if (!this.waitForTaskReplay) {
this.waitForTaskReplay = {
message,
attempt: 1,
idempotencyKey: randomUUID(),
};
} else {
if (
replayIdempotencyKey &&
replayIdempotencyKey !== this.waitForTaskReplay.idempotencyKey
) {
logger.error(
"wait for task handler called with mismatched idempotency key, won't overwrite replay request"
);
return;
throw new ExponentialBackoff.StopRetrying("No attempt ID");
}
this.waitForTaskReplay.attempt++;
return await this.#coordinatorSocket.socket.timeout(20_000).emitWithAck("WAIT_FOR_TASK", {
version: "v2",
friendlyId: message.friendlyId,
attemptFriendlyId: this.attemptFriendlyId,
});
});
if (!waitForTask.success) {
logger.error("Failed to wait for task with backoff", {
cause: waitForTask.cause,
error: waitForTask.error,
});
this.#emitUnrecoverableError(
"WaitForTaskFailed",
`${waitForTask.cause}: ${waitForTask.error}`
);
return;
}
}
const { willCheckpointAndRestore } = waitForTask.result;
await this.#prepareForWait("WAIT_FOR_TASK", willCheckpointAndRestore);
if (willCheckpointAndRestore) {
// We need to replay this on next connection if we don't receive RESUME_AFTER_DEPENDENCY within a reasonable time
if (!this.waitForTaskReplay) {
this.waitForTaskReplay = {
message,
attempt: 1,
idempotencyKey: randomUUID(),
};
} else {
if (
replayIdempotencyKey &&
replayIdempotencyKey !== this.waitForTaskReplay.idempotencyKey
) {
logger.error(
"wait for task handler called with mismatched idempotency key, won't overwrite replay request"
);
return;
}
this.waitForTaskReplay.attempt++;
}
}
};
}
// MARK: BATCH WAIT
async #waitForBatchHandler(message: OnWaitForBatchMessage, replayIdempotencyKey?: string) {
const waitForBatch = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Wait for batch with backoff", { retry });
#waitForBatchHandlerFactory(workerId?: string) {
return async (message: OnWaitForBatchMessage, replayIdempotencyKey?: string) => {
logger.log("onWaitForBatch", { workerId, message });
if (!this.attemptFriendlyId) {
logger.error("Failed to send wait message, attempt friendly ID not set", { message });
if (this.nextResumeAfter) {
logger.error("Already waiting for resume, skipping wait for batch", {
nextResumeAfter: this.nextResumeAfter,
});
throw new ExponentialBackoff.StopRetrying("No attempt ID");
return;
}
return await this.#coordinatorSocket.socket.timeout(20_000).emitWithAck("WAIT_FOR_BATCH", {
version: "v2",
batchFriendlyId: message.batchFriendlyId,
runFriendlyIds: message.runFriendlyIds,
attemptFriendlyId: this.attemptFriendlyId,
});
});
const waitForBatch = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Wait for batch with backoff", { retry });
if (!waitForBatch.success) {
logger.error("Failed to wait for batch with backoff", {
cause: waitForBatch.cause,
error: waitForBatch.error,
});
if (!this.attemptFriendlyId) {
logger.error("Failed to send wait message, attempt friendly ID not set", { message });
this.#emitUnrecoverableError(
"WaitForBatchFailed",
`${waitForBatch.cause}: ${waitForBatch.error}`
);
return;
}
const { willCheckpointAndRestore } = waitForBatch.result;
await this.#prepareForWait("WAIT_FOR_BATCH", willCheckpointAndRestore);
if (willCheckpointAndRestore) {
// We need to replay this on next connection if we don't receive RESUME_AFTER_DEPENDENCY within a reasonable time
if (!this.waitForBatchReplay) {
this.waitForBatchReplay = {
message,
attempt: 1,
idempotencyKey: randomUUID(),
};
} else {
if (
replayIdempotencyKey &&
replayIdempotencyKey !== this.waitForBatchReplay.idempotencyKey
) {
logger.error(
"wait for task handler called with mismatched idempotency key, won't overwrite replay request"
);
return;
throw new ExponentialBackoff.StopRetrying("No attempt ID");
}
this.waitForBatchReplay.attempt++;
return await this.#coordinatorSocket.socket.timeout(20_000).emitWithAck("WAIT_FOR_BATCH", {
version: "v2",
batchFriendlyId: message.batchFriendlyId,
runFriendlyIds: message.runFriendlyIds,
attemptFriendlyId: this.attemptFriendlyId,
});
});
if (!waitForBatch.success) {
logger.error("Failed to wait for batch with backoff", {
cause: waitForBatch.cause,
error: waitForBatch.error,
});
this.#emitUnrecoverableError(
"WaitForBatchFailed",
`${waitForBatch.cause}: ${waitForBatch.error}`
);
return;
}
}
const { willCheckpointAndRestore } = waitForBatch.result;
await this.#prepareForWait("WAIT_FOR_BATCH", willCheckpointAndRestore);
if (willCheckpointAndRestore) {
// We need to replay this on next connection if we don't receive RESUME_AFTER_DEPENDENCY within a reasonable time
if (!this.waitForBatchReplay) {
this.waitForBatchReplay = {
message,
attempt: 1,
idempotencyKey: randomUUID(),
};
} else {
if (
replayIdempotencyKey &&
replayIdempotencyKey !== this.waitForBatchReplay.idempotencyKey
) {
logger.error(
"wait for task handler called with mismatched idempotency key, won't overwrite replay request"
);
return;
}
this.waitForBatchReplay.attempt++;
}
}
};
}
// MARK: WORKER CREATION
#createBackgroundWorker() {
const workerId = randomUUID();
logger.log("Creating background worker", { workerId });
const backgroundWorker = new ProdBackgroundWorker("worker.js", {
projectConfig: __PROJECT_CONFIG__,
env: {
@@ -325,7 +347,10 @@ class ProdWorker {
});
backgroundWorker.onTaskHeartbeat.attach((attemptFriendlyId) => {
logger.log("onTaskHeartbeat", { attemptFriendlyId });
logger.log("onTaskHeartbeat", {
workerId,
attemptFriendlyId,
});
this.#coordinatorSocket.socket.volatile.emit("TASK_HEARTBEAT", {
version: "v1",
@@ -334,13 +359,19 @@ class ProdWorker {
});
backgroundWorker.onTaskRunHeartbeat.attach((runId) => {
logger.log("onTaskRunHeartbeat", { runId });
logger.log("onTaskRunHeartbeat", {
workerId,
runId,
});
this.#coordinatorSocket.socket.volatile.emit("TASK_RUN_HEARTBEAT", { version: "v1", runId });
});
backgroundWorker.onCreateTaskRunAttempt.attach(async (message) => {
logger.log("onCreateTaskRunAttempt()", { message });
logger.log("onCreateTaskRunAttempt()", {
workerId,
message,
});
const createAttempt = await defaultBackoff.execute(async ({ retry }) => {
logger.log("Create task run attempt with backoff", { retry });
@@ -377,6 +408,7 @@ class ProdWorker {
backgroundWorker.attemptCreatedNotification.attach((message) => {
logger.log("attemptCreatedNotification", {
workerId,
success: message.success,
...(message.success
? {
@@ -396,10 +428,24 @@ class ProdWorker {
// Workers with lazy attempt support set their friendly ID here
this.attemptFriendlyId = message.execution.attempt.id;
this.attemptNumber = message.execution.attempt.number;
});
// MARK: WAIT_FOR_DURATION
backgroundWorker.onWaitForDuration.attach(async (message) => {
logger.log("onWaitForDuration", { ...message, drift: Date.now() - message.now });
logger.log("onWaitForDuration", {
workerId,
...message,
drift: Date.now() - message.now,
});
if (this.nextResumeAfter) {
logger.error("Already waiting for resume, skipping wait for duration", {
nextResumeAfter: this.nextResumeAfter,
});
return;
}
noResume: {
const { ms, waitThresholdInMs } = message;
@@ -457,6 +503,17 @@ class ProdWorker {
// checkpointSafeInternalTimeout is accurate even after non-simulated restores
await Promise.race([internalTimeout, checkpointSafeInternalTimeout]);
const idempotencyKey = randomUUID();
this.durationResumeFallback = { idempotencyKey };
try {
await this.restoreNotification.waitFor(5_000);
} catch (error) {
logger.error("Did not receive restore notification in time", {
error,
});
}
try {
// The coordinator should cancel any in-progress checkpoints so we don't end up with race conditions
const { checkpointCanceled } = await this.#coordinatorSocket.socket
@@ -475,9 +532,6 @@ class ProdWorker {
logger.log("Waiting for external duration resume as we may have been restored");
const idempotencyKey = randomUUID();
this.durationResumeFallback = { idempotencyKey };
setTimeout(() => {
if (!this.durationResumeFallback) {
logger.error("Already resumed after duration, skipping fallback");
@@ -494,9 +548,12 @@ class ProdWorker {
this.#resumeAfterDuration();
}, 15_000);
} catch (error) {
// If the cancellation times out, we will proceed as if the checkpoint was canceled
logger.debug("Checkpoint cancellation timed out", { error });
break noResume;
// Just log this for now, but don't automatically resume. Wait for the external checkpoint-based resume.
logger.debug("Checkpoint cancellation timed out", {
workerId,
message,
error,
});
}
return;
@@ -505,8 +562,8 @@ class ProdWorker {
this.#resumeAfterDuration();
});
backgroundWorker.onWaitForTask.attach(this.#waitForTaskHandler.bind(this));
backgroundWorker.onWaitForBatch.attach(this.#waitForBatchHandler.bind(this));
backgroundWorker.onWaitForTask.attach(this.#waitForTaskHandlerFactory(workerId).bind(this));
backgroundWorker.onWaitForBatch.attach(this.#waitForBatchHandlerFactory(workerId).bind(this));
return backgroundWorker;
}
@@ -514,6 +571,18 @@ class ProdWorker {
async #prepareForWait(reason: WaitReason, willCheckpointAndRestore: boolean) {
logger.log(`prepare for ${reason}`, { willCheckpointAndRestore });
if (this.nextResumeAfter) {
logger.error("Already waiting for resume, skipping prepare for wait", {
nextResumeAfter: this.nextResumeAfter,
params: {
reason,
willCheckpointAndRestore,
},
});
return;
}
if (!willCheckpointAndRestore) {
return;
}
@@ -526,19 +595,11 @@ class ProdWorker {
}
// MARK: RETRY PREP
async #prepareForRetry(
willCheckpointAndRestore: boolean,
shouldExit: boolean,
exitCode?: number
) {
logger.log("prepare for retry", { willCheckpointAndRestore, shouldExit, exitCode });
async #prepareForRetry(shouldExit: boolean, exitCode?: number) {
logger.log("prepare for retry", { shouldExit, exitCode });
// Graceful shutdown on final attempt
if (shouldExit) {
if (willCheckpointAndRestore) {
logger.error("WARNING: Will checkpoint but also requested exit. This won't end well.");
}
await this.#exitGracefully(false, exitCode);
return;
}
@@ -548,15 +609,7 @@ class ProdWorker {
this.waitForPostStart = false;
this.executing = false;
this.attemptFriendlyId = undefined;
if (!willCheckpointAndRestore) {
return;
}
this.waitForPostStart = true;
// We already flush after completion, so we don't need to do it here
await this.#prepareForCheckpoint(false);
this.attemptNumber = undefined;
}
// MARK: CHECKPOINT PREP
@@ -591,6 +644,8 @@ class ProdWorker {
this.nextResumeAfter = undefined;
this.waitForPostStart = false;
this.durationResumeFallback = undefined;
this.#backgroundWorker.waitCompletedNotification();
}
@@ -683,7 +738,7 @@ class ProdWorker {
return await this.#coordinatorSocket.socket
.timeout(20_000)
.emitWithAck("TASK_RUN_COMPLETED", {
version: "v1",
version: "v2",
execution,
completion,
});
@@ -711,32 +766,10 @@ class ProdWorker {
? EXIT_CODE_CHILD_NONZERO
: 0;
await this.#prepareForRetry(willCheckpointAndRestore, shouldExit, exitCode);
await this.#prepareForRetry(shouldExit, exitCode);
if (willCheckpointAndRestore) {
// We need to replay this on next connection if we don't receive READY_FOR_RETRY within a reasonable time
if (!this.submitAttemptCompletionReplay) {
this.submitAttemptCompletionReplay = {
message: {
execution,
completion,
},
attempt: 1,
idempotencyKey: randomUUID(),
};
} else {
if (
replayIdempotencyKey &&
replayIdempotencyKey !== this.submitAttemptCompletionReplay.idempotencyKey
) {
logger.error(
"attempt completion handler called with mismatched idempotency key, won't overwrite replay request"
);
return;
}
this.submitAttemptCompletionReplay.attempt++;
}
logger.error("This worker should never be checkpointed between attempts. This is a bug.");
}
}
@@ -767,6 +800,10 @@ class ProdWorker {
extraHeaders["x-trigger-attempt-friendly-id"] = this.attemptFriendlyId;
}
if (this.attemptNumber !== undefined) {
extraHeaders["x-trigger-attempt-number"] = String(this.attemptNumber);
}
logger.log(`connecting to coordinator: ${host}:${COORDINATOR_PORT}`);
logger.debug(`connecting with extra headers`, { extraHeaders });
@@ -851,40 +888,14 @@ class ProdWorker {
return;
}
this.durationResumeFallback = undefined;
this.#resumeAfterDuration();
},
// Deprecated: This will never get called as this worker supports lazy attempts. It's only here for a quick view of the flow old workers use.
EXECUTE_TASK_RUN: async ({ executionPayload }) => {
if (this.executing) {
logger.error("dropping execute request, already executing");
return;
}
if (this.completed.has(executionPayload.execution.attempt.id)) {
logger.error("dropping execute request, already completed");
return;
}
this.executing = true;
this.attemptFriendlyId = executionPayload.execution.attempt.id;
const completion = await this.#backgroundWorker.executeTaskRun(executionPayload);
logger.log("completed", completion);
this.completed.add(executionPayload.execution.attempt.id);
const { willCheckpointAndRestore, shouldExit } =
await this.#coordinatorSocket.socket.emitWithAck("TASK_RUN_COMPLETED", {
version: "v1",
execution: executionPayload.execution,
completion,
});
logger.log("completion acknowledged", { willCheckpointAndRestore, shouldExit });
await this.#prepareForRetry(willCheckpointAndRestore, shouldExit);
EXECUTE_TASK_RUN: async () => {
// These messages should only be received by old workers that don't support lazy attempts
this.#failRun(
this.runId,
"Received deprecated EXECUTE_TASK_RUN message. Please contact us if you see this error."
);
},
EXECUTE_TASK_RUN_LAZY_ATTEMPT: async (message) => {
this.readyForLazyAttemptReplay = undefined;
@@ -947,8 +958,6 @@ class ProdWorker {
return;
}
this.submitAttemptCompletionReplay = undefined;
await this.#readyForLazyAttempt();
},
},
@@ -960,7 +969,11 @@ class ProdWorker {
});
// We need to send our current state to the coordinator
socket.emit("SET_STATE", { version: "v1", attemptFriendlyId: this.attemptFriendlyId });
socket.emit("SET_STATE", {
version: "v1",
attemptFriendlyId: this.attemptFriendlyId,
attemptNumber: this.attemptNumber ? String(this.attemptNumber) : undefined,
});
try {
if (this.waitForPostStart) {
@@ -981,7 +994,7 @@ class ProdWorker {
}
if (!this.attemptFriendlyId) {
logger.error("Missing friendly ID", { status: this.#status });
logger.error("Missing attempt friendly ID", { status: this.#status });
this.#emitUnrecoverableError(
"NoAttemptId",
@@ -991,9 +1004,21 @@ class ProdWorker {
return;
}
if (!this.attemptNumber) {
logger.error("Missing attempt number", { status: this.#status });
this.#emitUnrecoverableError(
"NoAttemptNumber",
"Attempt number not set while resuming from paused state"
);
return;
}
socket.emit("READY_FOR_RESUME", {
version: "v1",
version: "v2",
attemptFriendlyId: this.attemptFriendlyId,
attemptNumber: this.attemptNumber,
type: this.nextResumeAfter,
});
@@ -1169,7 +1194,7 @@ class ProdWorker {
try {
await backoff.wait(attempt + 1);
await this.#waitForTaskHandler(message);
await this.#waitForTaskHandlerFactory("replay")(message, idempotencyKey);
} catch (error) {
if (error instanceof ExponentialBackoff.RetryLimitExceeded) {
logger.error("wait for task replay retry limit exceeded", { error });
@@ -1212,7 +1237,7 @@ class ProdWorker {
try {
await backoff.wait(attempt + 1);
await this.#waitForBatchHandler(message);
await this.#waitForBatchHandlerFactory("replay")(message, idempotencyKey);
} catch (error) {
if (error instanceof ExponentialBackoff.RetryLimitExceeded) {
logger.error("wait for batch replay retry limit exceeded", { error });
@@ -1223,49 +1248,6 @@ class ProdWorker {
return;
}
if (this.submitAttemptCompletionReplay) {
logger.log("replaying attempt completion", {
...this.submitAttemptCompletionReplay,
cancellationDelay: replayCancellationDelay,
});
const { idempotencyKey, message, attempt } = this.submitAttemptCompletionReplay;
// Give the platform some time to send READY_FOR_RETRY
await timeout(replayCancellationDelay);
if (!this.submitAttemptCompletionReplay) {
logger.error("attempt completion replay cancelled, discarding", {
originalMessage: { idempotencyKey, message, attempt },
});
return;
}
if (idempotencyKey !== this.submitAttemptCompletionReplay.idempotencyKey) {
logger.error("attempt completion replay idempotency key mismatch, discarding", {
originalMessage: { idempotencyKey, message, attempt },
newMessage: this.submitAttemptCompletionReplay,
});
return;
}
try {
await backoff.wait(attempt + 1);
await this.#submitAttemptCompletion(message.execution, message.completion, idempotencyKey);
} catch (error) {
if (error instanceof ExponentialBackoff.RetryLimitExceeded) {
logger.error("attempt completion replay retry limit exceeded", { error });
} else {
logger.error("attempt completion replay error", { error });
}
}
return;
}
}
// MARK: HTTP SERVER
@@ -1345,6 +1327,7 @@ class ProdWorker {
}
case "restore": {
await this.#reconnectAfterPostStart();
this.restoreNotification.post();
break;
}
default: {
@@ -1452,6 +1435,7 @@ class ProdWorker {
nextResumeAfter: this.nextResumeAfter,
waitForPostStart: this.waitForPostStart,
attemptFriendlyId: this.attemptFriendlyId,
attemptNumber: this.attemptNumber,
waitForTaskReplay: this.waitForTaskReplay,
waitForBatchReplay: this.waitForBatchReplay,
};
packages/core-apps/src/provider.ts
+4
View File
@@ -43,6 +43,7 @@ export interface TaskOperationsCreateOptions {
image: string;
machine: MachinePreset;
version: string;
nextAttemptNumber?: number;
// identifiers
envId: string;
envType: EnvironmentType;
@@ -55,6 +56,7 @@ export interface TaskOperationsRestoreOptions {
imageRef: string;
checkpointRef: string;
machine: MachinePreset;
attemptNumber?: number;
// identifiers
envId: string;
envType: EnvironmentType;
@@ -142,6 +144,7 @@ export class ProviderShell implements Provider {
image: message.data.image,
machine: message.data.machine,
version: message.data.version,
nextAttemptNumber: message.data.nextAttemptNumber,
// identifiers
envId: message.data.envId,
envType: message.data.envType,
@@ -278,6 +281,7 @@ export class ProviderShell implements Provider {
checkpointRef: message.location,
machine: message.machine,
imageRef: message.imageRef,
attemptNumber: message.attemptNumber,
// identifiers
envId: message.envId,
envType: message.envType,
packages/core/src/v3/schemas/messages.ts
+19 -7
View File
@@ -31,6 +31,7 @@ export const BackgroundWorkerServerMessages = z.discriminatedUnion("type", [
image: z.string(),
version: z.string(),
machine: MachinePreset,
nextAttemptNumber: z.number().optional(),
// identifiers
id: z.string().optional(), // TODO: Remove this completely in a future release
envId: z.string(),
@@ -357,6 +358,7 @@ export const PlatformToProviderMessages = {
location: z.string(),
reason: z.string().optional(),
imageRef: z.string(),
attemptNumber: z.number().optional(),
machine: MachinePreset,
// identifiers
checkpointId: z.string(),
@@ -482,7 +484,7 @@ export const CoordinatorToPlatformMessages = {
},
TASK_RUN_COMPLETED: {
message: z.object({
version: z.literal("v1").default("v1"),
version: z.enum(["v1", "v2"]).default("v1"),
execution: ProdTaskRunExecution,
completion: TaskRunExecutionResult,
checkpoint: z
@@ -701,11 +703,19 @@ export const ProdWorkerToCoordinatorMessages = {
}),
},
READY_FOR_RESUME: {
message: z.object({
version: z.literal("v1").default("v1"),
attemptFriendlyId: z.string(),
type: WaitReason,
}),
message: z.discriminatedUnion("version", [
z.object({
version: z.literal("v1"),
attemptFriendlyId: z.string(),
type: WaitReason,
}),
z.object({
version: z.literal("v2"),
attemptFriendlyId: z.string(),
attemptNumber: z.number(),
type: WaitReason,
}),
]),
},
READY_FOR_CHECKPOINT: {
message: z.object({
@@ -744,7 +754,7 @@ export const ProdWorkerToCoordinatorMessages = {
},
TASK_RUN_COMPLETED: {
message: z.object({
version: z.literal("v1").default("v1"),
version: z.enum(["v1", "v2"]).default("v1"),
execution: ProdTaskRunExecution,
completion: TaskRunExecutionResult,
}),
@@ -835,6 +845,7 @@ export const ProdWorkerToCoordinatorMessages = {
message: z.object({
version: z.literal("v1").default("v1"),
attemptFriendlyId: z.string().optional(),
attemptNumber: z.string().optional(),
}),
},
};
@@ -899,6 +910,7 @@ export const ProdWorkerSocketData = z.object({
envId: z.string(),
runId: z.string(),
attemptFriendlyId: z.string().optional(),
attemptNumber: z.string().optional(),
podName: z.string(),
deploymentId: z.string(),
deploymentVersion: z.string(),
packages/database/prisma/migrations/20240809213411_add_checkpoint_attempt_number_column/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "Checkpoint" ADD COLUMN "attemptNumber" INTEGER;
packages/database/prisma/schema.prisma
+3 -2
View File
@@ -2115,8 +2115,9 @@ model Checkpoint {
run TaskRun @relation(fields: [runId], references: [id], onDelete: Cascade, onUpdate: Cascade)
runId String
attempt TaskRunAttempt @relation(fields: [attemptId], references: [id], onDelete: Cascade, onUpdate: Cascade)
attemptId String
attempt TaskRunAttempt @relation(fields: [attemptId], references: [id], onDelete: Cascade, onUpdate: Cascade)
attemptId String
attemptNumber Int?
project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade)
projectId String