Frozen run fixes (#1286)

* When resuming a batch, only do marqs operations once

* Made TaskRunDependency clearer in the Prisma schema

* New ResumeDependentParentsService service, use it from checkpoints

* WIP on making resuming more robust

* Turn the declarative schedules off because they make debugging other runs painful

* Resuming batches when there’s an attempt is working

* If there’s no attempt then create one

* Added a log if there are no span events to complete

* If Graphile addJob doesn’t return a row, log and return undefined. No throw

* Pass prisma into the ResumeDependentParentsService

* Removed the todos

* Pass Prisma through to the checkpoint service

* Fix for not checking the batch item correctly

* Fix for when a log flush times out and the process is checkpointed

* Fix for when a log flush times out and the process is checkpointed

* Another test run that does batches with failed subtasks

* Don’t call ResumeTaskRunDependenciesService anymore (we have a new service)

* Only resume if the run is in a final state

* If an attempt doesn’t exist, fix for creating queue with sanitized name

* If DEV then don’t resume using marqs/batches. The CLI manages it

* We don’t need to check the run status again, it’s in the main function now

* Added TaskRunAttempt taskRunId index

* Only allow calling ResumeDependentParentsService with a run ID

* Put the flushing back to what it was

Author: matt-aitken

.changeset/tidy-pets-smell.md

@@ -0,0 +1,6 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Fix for when a log flush times out and the process is checkpointed

apps/webapp/app/platform/zodWorker.server.ts

@@ -318,7 +318,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
     identifier: K,
     payload: z.infer<TMessageCatalog[K]>,
     options?: ZodWorkerEnqueueOptions
-  ): Promise<GraphileJob> {
+  ): Promise<GraphileJob | undefined> {
     const task = this.#tasks[identifier];
 
     const optionsWithoutTx = removeUndefinedKeys(omit(options ?? {}, ["tx"]));
@@ -439,11 +439,9 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
         identifier,
         payload,
         spec,
+        error: JSON.stringify(rows.error),
       });
-
-      throw new Error(
-        `Failed to add job to queue, zod parsing error: ${JSON.stringify(rows.error)}`
-      );
+      return { job: undefined, durationInMs: Math.floor(durationInMs) };
     }
 
     const job = rows.data[0];

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -796,6 +796,12 @@ function RunTimelineLine({ title, state }: RunTimelineLineProps) {
 function RunError({ error }: { error: TaskRunError }) {
   switch (error.type) {
     case "STRING_ERROR":
+      return (
+        <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">Error</Header3>
+          <Callout variant="error">{error.raw}</Callout>
+        </div>
+      );
     case "CUSTOM_ERROR": {
       return (
         <div className="flex flex-col gap-2 rounded-sm border border-rose-500/50 px-3 pb-3 pt-2">

apps/webapp/app/services/schedules/deliverScheduledEvent.server.ts

@@ -134,7 +134,7 @@ export class DeliverScheduledEventService {
           id,
         },
         data: {
-          workerJobId: workerJob.id,
+          workerJobId: workerJob?.id,
           nextEventTimestamp: runAt,
         },
       });

apps/webapp/app/v3/eventRepository.server.ts

@@ -226,6 +226,7 @@ export class EventRepository {
     const events = await this.queryIncompleteEvents({ spanId });
 
     if (events.length === 0) {
+      logger.warn("No incomplete events found for spanId", { spanId, options });
       return;
     }
 

apps/webapp/app/v3/failedTaskRun.server.ts

@@ -42,26 +42,10 @@ export class FailedTaskRunService extends BaseService {
       id: taskRun.id,
       status: "SYSTEM_FAILURE",
       completedAt: new Date(),
+      attemptStatus: "FAILED",
+      error: sanitizeError(completion.error),
     });
 
-    // Get the final attempt and add the error to it, if it's not already set
-    const finalAttempt = await this._prisma.taskRunAttempt.findFirst({
-      where: {
-        taskRunId: taskRun.id,
-      },
-      orderBy: { id: "desc" },
-    });
-
-    if (finalAttempt && !finalAttempt.error) {
-      // Haven't set the status because the attempt might still be running
-      await this._prisma.taskRunAttempt.update({
-        where: { id: finalAttempt.id },
-        data: {
-          error: sanitizeError(completion.error),
-        },
-      });
-    }
-
     // Now we need to "complete" the task run event/span
     await eventRepository.completeEvent(taskRun.spanId, {
       endTime: new Date(),

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

@@ -5,7 +5,6 @@ import { eventRepository } from "../eventRepository.server";
 import { isCancellableRunStatus } from "../taskStatus";
 import { BaseService } from "./baseService.server";
 import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
-import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server";
 
 export class CancelAttemptService extends BaseService {
   public async call(
@@ -61,13 +60,15 @@ export class CancelAttemptService extends BaseService {
           },
         });
 
+        const isCancellable = isCancellableRunStatus(taskRunAttempt.taskRun.status);
+
         const finalizeService = new FinalizeTaskRunService(tx);
         await finalizeService.call({
           id: taskRunId,
-          status: isCancellableRunStatus(taskRunAttempt.taskRun.status) ? "INTERRUPTED" : undefined,
-          completedAt: isCancellableRunStatus(taskRunAttempt.taskRun.status)
-            ? cancelledAt
-            : undefined,
+          status: isCancellable ? "INTERRUPTED" : undefined,
+          completedAt: isCancellable ? cancelledAt : undefined,
+          attemptStatus: isCancellable ? "CANCELED" : undefined,
+          error: isCancellable ? { type: "STRING_ERROR", raw: reason } : undefined,
         });
       });
 
@@ -84,10 +85,6 @@ export class CancelAttemptService extends BaseService {
           return eventRepository.cancelEvent(event, cancelledAt, reason);
         })
       );
-
-      if (environment?.type !== "DEVELOPMENT") {
-        await ResumeTaskRunDependenciesService.enqueue(taskRunAttempt.id, this._prisma);
-      }
     });
   }
 }

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

@@ -76,6 +76,11 @@ export class CancelTaskRunService extends BaseService {
         runtimeEnvironment: true,
         lockedToVersion: true,
       },
+      attemptStatus: "CANCELED",
+      error: {
+        type: "STRING_ERROR",
+        raw: opts.reason,
+      },
     });
 
     const inProgressEvents = await eventRepository.queryIncompleteEvents({

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

@@ -17,7 +17,6 @@ import { createExceptionPropertiesFromError, eventRepository } from "../eventRep
 import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { CancelAttemptService } from "./cancelAttempt.server";
-import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server";
 import { MAX_TASK_RUN_ATTEMPTS } from "~/consts";
 import { CreateCheckpointService } from "./createCheckpoint.server";
 import { TaskRun } from "@trigger.dev/database";
@@ -76,6 +75,12 @@ export class CompleteAttemptService extends BaseService {
         id: run.id,
         status: "SYSTEM_FAILURE",
         completedAt: new Date(),
+        attemptStatus: "FAILED",
+        error: {
+          type: "INTERNAL_ERROR",
+          code: "TASK_EXECUTION_FAILED",
+          message: "Tried to complete attempt but it doesn't exist",
+        },
       });
 
       // No attempt, so there's no message to ACK
@@ -149,10 +154,6 @@ export class CompleteAttemptService extends BaseService {
       },
     });
 
-    if (!env || env.type !== "DEVELOPMENT") {
-      await ResumeTaskRunDependenciesService.enqueue(taskRunAttempt.id, this._prisma);
-    }
-
     return "COMPLETED";
   }
 
@@ -355,10 +356,6 @@ export class CompleteAttemptService extends BaseService {
         });
       }
 
-      if (!env || env.type !== "DEVELOPMENT") {
-        await ResumeTaskRunDependenciesService.enqueue(taskRunAttempt.id, this._prisma);
-      }
-
       return "COMPLETED";
     }
   }

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

@@ -4,7 +4,6 @@ import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { logger } from "~/services/logger.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server";
 import { CRASHABLE_ATTEMPT_STATUSES, isCrashableRunStatus } from "../taskStatus";
 import { sanitizeError } from "@trigger.dev/core/v3";
 import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
@@ -69,6 +68,13 @@ export class CrashTaskRunService extends BaseService {
           },
         },
       },
+      attemptStatus: "FAILED",
+      error: {
+        type: "INTERNAL_ERROR",
+        code: "TASK_RUN_CRASHED",
+        message: opts.reason,
+        stackTrace: opts.logs,
+      },
     });
 
     const inProgressEvents = await eventRepository.queryIncompleteEvents(
@@ -146,12 +152,6 @@ export class CrashTaskRunService extends BaseService {
           }),
         },
       });
-
-      if (environment.type === "DEVELOPMENT") {
-        return;
-      }
-
-      await ResumeTaskRunDependenciesService.enqueue(attempt.id, this._prisma);
     });
   }
 }

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

@@ -4,16 +4,11 @@ import type { Checkpoint, CheckpointRestoreEvent } from "@trigger.dev/database";
 import { logger } from "~/services/logger.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import {
-  isFinalAttemptStatus,
-  isFinalRunStatus,
-  isFreezableAttemptStatus,
-  isFreezableRunStatus,
-} from "../taskStatus";
+import { isFreezableAttemptStatus, isFreezableRunStatus } from "../taskStatus";
 import { BaseService } from "./baseService.server";
 import { CreateCheckpointRestoreEventService } from "./createCheckpointRestoreEvent.server";
 import { ResumeBatchRunService } from "./resumeBatchRun.server";
-import { ResumeTaskDependencyService } from "./resumeTaskDependency.server";
+import { ResumeDependentParentsService } from "./resumeDependentParents.server";
 
 export class CreateCheckpointService extends BaseService {
   public async call(
@@ -177,127 +172,15 @@ export class CreateCheckpointService extends BaseService {
           });
           await marqs?.cancelHeartbeat(attempt.taskRunId);
 
-          const dependency = await this._prisma.taskRunDependency.findFirst({
-            select: {
-              id: true,
-              taskRunId: true,
-            },
-            where: {
-              taskRun: {
-                friendlyId: reason.friendlyId,
-              },
-            },
-          });
-
-          logger.log("CreateCheckpointService: Created checkpoint WAIT_FOR_TASK", {
-            checkpointId: checkpoint.id,
-            runFriendlyId: reason.friendlyId,
-            dependencyId: dependency?.id,
-          });
-
-          if (!dependency) {
-            logger.error("CreateCheckpointService: Dependency not found", {
-              friendlyId: reason.friendlyId,
-            });
-
-            return {
-              success: true,
-              checkpoint,
-              event: checkpointEvent,
-              keepRunAlive: false,
-            };
-          }
-
-          const childRun = await this._prisma.taskRun.findFirst({
-            select: {
-              id: true,
-              status: true,
-            },
-            where: {
-              id: dependency.taskRunId,
-            },
-          });
-
-          if (!childRun) {
-            logger.error("CreateCheckpointService: Dependency child run not found", {
-              taskRunId: dependency.taskRunId,
-              runFriendlyId: reason.friendlyId,
-              dependencyId: dependency.id,
-            });
-
-            return {
-              success: true,
-              checkpoint,
-              event: checkpointEvent,
-              keepRunAlive: false,
-            };
-          }
+          const resumeService = new ResumeDependentParentsService(this._prisma);
+          const result = await resumeService.call({ id: attempt.taskRunId });
 
-          const isFinished = isFinalRunStatus(childRun.status);
-          if (!isFinished) {
-            logger.debug("CreateCheckpointService: Dependency child run not finished", {
-              taskRunId: dependency.taskRunId,
-              runFriendlyId: reason.friendlyId,
-              dependencyId: dependency.id,
-              childRunStatus: childRun.status,
-              childRunId: childRun.id,
-            });
-
-            return {
-              success: true,
-              checkpoint,
-              event: checkpointEvent,
-              keepRunAlive: false,
-            };
-          }
-
-          const lastAttempt = await this._prisma.taskRunAttempt.findFirst({
-            select: {
-              id: true,
-              status: true,
-            },
-            where: {
-              taskRunId: dependency.taskRunId,
-            },
-            orderBy: {
-              createdAt: "desc",
-            },
-          });
-
-          if (!lastAttempt) {
-            logger.debug("CreateCheckpointService: Dependency child attempt not found", {
-              taskRunId: dependency.taskRunId,
-              runFriendlyId: reason.friendlyId,
-              dependencyId: dependency?.id,
-            });
-            return {
-              success: true,
-              checkpoint,
-              event: checkpointEvent,
-              keepRunAlive: false,
-            };
-          }
-
-          if (!isFinalAttemptStatus(lastAttempt.status)) {
-            logger.debug("CreateCheckpointService: Dependency child attempt not final", {
-              taskRunId: dependency.taskRunId,
-              runFriendlyId: reason.friendlyId,
-              dependencyId: dependency.id,
-              lastAttemptId: lastAttempt.id,
-              lastAttemptStatus: lastAttempt.status,
-            });
-
-            return {
-              success: true,
-              checkpoint,
-              event: checkpointEvent,
-              keepRunAlive: false,
-            };
+          if (result.success) {
+            logger.log("CreateCheckpointService: Resumed dependent parents", result);
+          } else {
+            logger.error("CreateCheckpointService: Failed to resume dependent parents", result);
           }
 
-          //resume the dependent task
-          await ResumeTaskDependencyService.enqueue(dependency.id, lastAttempt.id, this._prisma);
-
           return {
             success: true,
             checkpoint,

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

@@ -45,6 +45,11 @@ export class ExpireEnqueuedRunService extends BaseService {
       status: "EXPIRED",
       expiredAt: new Date(),
       completedAt: new Date(),
+      attemptStatus: "FAILED",
+      error: {
+        type: "STRING_ERROR",
+        raw: `Run expired because the TTL (${run.ttl}) was reached`,
+      },
     });
 
     await eventRepository.completeEvent(run.spanId, {