handle reserve concurrency with recursive deadlocks

Author: ericallam

apps/webapp/app/components/runs/v3/BatchStatus.tsx

@@ -1,16 +1,17 @@
-import { CheckCircleIcon } from "@heroicons/react/20/solid";
+import { CheckCircleIcon, XCircleIcon } from "@heroicons/react/20/solid";
 import { BatchTaskRunStatus } from "@trigger.dev/database";
 import assertNever from "assert-never";
 import { Spinner } from "~/components/primitives/Spinner";
 import { cn } from "~/utils/cn";
 
-export const allBatchStatuses = ["PENDING", "COMPLETED"] as const satisfies Readonly<
+export const allBatchStatuses = ["PENDING", "COMPLETED", "ABORTED"] as const satisfies Readonly<
   Array<BatchTaskRunStatus>
 >;
 
 const descriptions: Record<BatchTaskRunStatus, string> = {
   PENDING: "The batch has child runs that have not yet completed.",
   COMPLETED: "All the batch child runs have finished.",
+  ABORTED: "The batch was aborted because some child tasks could not be triggered.",
 };
 
 export function descriptionForBatchStatus(status: BatchTaskRunStatus): string {
@@ -50,6 +51,8 @@ export function BatchStatusIcon({
       return <Spinner className={cn(batchStatusColor(status), className)} />;
     case "COMPLETED":
       return <CheckCircleIcon className={cn(batchStatusColor(status), className)} />;
+    case "ABORTED":
+      return <XCircleIcon className={cn(batchStatusColor(status), className)} />;
     default: {
       assertNever(status);
     }
@@ -62,6 +65,8 @@ export function batchStatusColor(status: BatchTaskRunStatus): string {
       return "text-pending";
     case "COMPLETED":
       return "text-success";
+    case "ABORTED":
+      return "text-error";
     default: {
       assertNever(status);
     }
@@ -74,6 +79,8 @@ export function batchStatusTitle(status: BatchTaskRunStatus): string {
       return "In progress";
     case "COMPLETED":
       return "Completed";
+    case "ABORTED":
+      return "Aborted";
     default: {
       assertNever(status);
     }

apps/webapp/app/presenters/v3/BatchListPresenter.server.ts

@@ -190,7 +190,7 @@ WHERE
           throw new Error(`Environment not found for Batch ${batch.id}`);
         }
 
-        const hasFinished = batch.status === "COMPLETED";
+        const hasFinished = batch.status !== "PENDING";
 
         return {
           id: batch.id,

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -118,6 +118,9 @@ export class SpanPresenter extends BasePresenter {
         metadata: true,
         metadataType: true,
         maxAttempts: true,
+        output: true,
+        outputType: true,
+        error: true,
         project: {
           include: {
             organization: true,
@@ -162,31 +165,13 @@ export class SpanPresenter extends BasePresenter {
 
     const isFinished = isFinalRunStatus(run.status);
 
-    const finishedAttempt = isFinished
-      ? await this._replica.taskRunAttempt.findFirst({
-          select: {
-            output: true,
-            outputType: true,
-            error: true,
-          },
-          where: {
-            status: { in: FINAL_ATTEMPT_STATUSES },
-            taskRunId: run.id,
-          },
-          orderBy: {
-            createdAt: "desc",
-          },
-        })
-      : null;
-
-    const output =
-      finishedAttempt === null
-        ? undefined
-        : finishedAttempt.outputType === "application/store"
-        ? `/resources/packets/${run.runtimeEnvironment.id}/${finishedAttempt.output}`
-        : typeof finishedAttempt.output !== "undefined" && finishedAttempt.output !== null
-        ? await prettyPrintPacket(finishedAttempt.output, finishedAttempt.outputType ?? undefined)
-        : undefined;
+    const output = !isFinished
+      ? undefined
+      : run.outputType === "application/store"
+      ? `/resources/packets/${run.runtimeEnvironment.id}/${run.output}`
+      : typeof run.output !== "undefined" && run.output !== null
+      ? await prettyPrintPacket(run.output, run.outputType ?? undefined)
+      : undefined;
 
     const payload =
       run.payloadType === "application/store"
@@ -196,14 +181,14 @@ export class SpanPresenter extends BasePresenter {
         : undefined;
 
     let error: TaskRunError | undefined = undefined;
-    if (finishedAttempt?.error) {
-      const result = TaskRunError.safeParse(finishedAttempt.error);
+    if (run?.error) {
+      const result = TaskRunError.safeParse(run.error);
       if (result.success) {
         error = result.data;
       } else {
         error = {
           type: "CUSTOM_ERROR",
-          raw: JSON.stringify(finishedAttempt.error),
+          raw: JSON.stringify(run.error),
         };
       }
     }
@@ -300,7 +285,7 @@ export class SpanPresenter extends BasePresenter {
       payload,
       payloadType: run.payloadType,
       output,
-      outputType: finishedAttempt?.outputType ?? "application/json",
+      outputType: run?.outputType ?? "application/json",
       error,
       relationships: {
         root: run.rootTaskRun

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

@@ -95,6 +95,8 @@ export type EventBuilder = {
   traceId: string;
   spanId: string;
   setAttribute: SetAttribute<TraceAttributes>;
+  stop: () => void;
+  failWithError: (error: TaskRunError) => void;
 };
 
 export type EventRepoConfig = {
@@ -916,6 +918,9 @@ export class EventRepository {
           ]
         : [];
 
+    let isStopped = false;
+    let failedWithError: TaskRunError | undefined;
+
     const eventBuilder = {
       traceId,
       spanId,
@@ -933,10 +938,20 @@ export class EventRepository {
           }
         }
       },
+      stop: () => {
+        isStopped = true;
+      },
+      failWithError: (error: TaskRunError) => {
+        failedWithError = error;
+      },
     };
 
     const result = await callback(eventBuilder, traceContext, propagatedContext?.traceparent);
 
+    if (isStopped) {
+      return result;
+    }
+
     const duration = process.hrtime.bigint() - start;
 
     const metadata = {
@@ -970,13 +985,14 @@ export class EventRepository {
       parentId,
       tracestate,
       duration: options.incomplete ? 0 : duration,
-      isPartial: options.incomplete,
+      isPartial: failedWithError ? false : options.incomplete,
+      isError: !!failedWithError,
       message: message,
       serviceName: "api server",
       serviceNamespace: "trigger.dev",
       level: "TRACE",
       kind: options.kind,
-      status: "OK",
+      status: failedWithError ? "ERROR" : "OK",
       startTime,
       environmentId: options.environment.id,
       environmentType: options.environment.type,
@@ -1004,6 +1020,17 @@ export class EventRepository {
       payload: options.attributes.payload,
       payloadType: options.attributes.payloadType,
       idempotencyKey: options.attributes.idempotencyKey,
+      events: failedWithError
+        ? [
+            {
+              name: "exception",
+              time: new Date(),
+              properties: {
+                exception: createExceptionPropertiesFromError(failedWithError),
+              },
+            },
+          ]
+        : undefined,
     };
 
     if (options.immediate) {