If a checkpoint has been created, the coordinator won’t continue the …

…run with RESUME_AFTER_DEPENDENCY_WITH_ACK
triggerdotdev · Sep 17, 2024 · 2dd5d46 · 2dd5d46
1 parent 7256244
commit 2dd5d46
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 0 deletions.
diff --git a/apps/coordinator/src/index.ts b/apps/coordinator/src/index.ts
@@ -178,6 +178,22 @@ class TaskCoordinator {
             };
           }
 
+          //if this is set, we want to kill the process because it will be resumed with the checkpoint from the queue
+          if (taskSocket.data.requiresCheckpointResumeWithMessage) {
+            logger.log("RESUME_AFTER_DEPENDENCY_WITH_ACK: Checkpoint is set so going to nack", {
+              socketData: taskSocket.data,
+            });
+
+            return {
+              success: false,
+              error: {
+                name: "CheckpointMessagePresentError",
+                message:
+                  "Checkpoint message is present, so we need to kill the process and resume from the queue.",
+              },
+            };
+          }
+
           await chaosMonkey.call();
 
           // In case the task resumed faster than we could checkpoint
@@ -819,6 +835,12 @@ class TaskCoordinator {
             return;
           }
 
+          //setting this means we can only resume from a checkpoint
+          socket.data.requiresCheckpointResumeWithMessage = `location:${checkpoint.location}-docker:${checkpoint.docker}`;
+          logger.log("WAIT_FOR_TASK set requiresCheckpointResumeWithMessage", {
+            requiresCheckpointResumeWithMessage: socket.data.requiresCheckpointResumeWithMessage,
+          });
+
           const ack = await this.#platformSocket?.sendWithAck("CHECKPOINT_CREATED", {
             version: "v1",
             attemptFriendlyId: message.attemptFriendlyId,
@@ -889,6 +911,12 @@ class TaskCoordinator {
             return;
           }
 
+          //setting this means we can only resume from a checkpoint
+          socket.data.requiresCheckpointResumeWithMessage = `location:${checkpoint.location}-docker:${checkpoint.docker}`;
+          logger.log("WAIT_FOR_BATCH set requiresCheckpointResumeWithMessage", {
+            requiresCheckpointResumeWithMessage: socket.data.requiresCheckpointResumeWithMessage,
+          });
+
           const ack = await this.#platformSocket?.sendWithAck("CHECKPOINT_CREATED", {
             version: "v1",
             attemptFriendlyId: message.attemptFriendlyId,

diff --git a/packages/core/src/v3/schemas/messages.ts b/packages/core/src/v3/schemas/messages.ts
@@ -861,6 +861,7 @@ export const ProdWorkerSocketData = z.object({
   podName: z.string(),
   deploymentId: z.string(),
   deploymentVersion: z.string(),
+  requiresCheckpointResumeWithMessage: z.string().optional(),
 });
 
 export const CoordinatorSocketData = z.object({