handle dependency resume edge case

Author: nicktrn

apps/webapp/app/v3/services/createCheckpoint.server.ts

@@ -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 {

apps/webapp/app/v3/services/resumeBatchRun.server.ts

@@ -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,
@@ -84,6 +75,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

@@ -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,
@@ -61,6 +52,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],