v3: Fixes for using (batch)triggerAndWait with idempotency keys (#1043)

* Fixes various issues with triggerAndWait and batchTriggerAndWait

When using idempotency keys, triggerAndWait and batchTriggerAndWait will still work even if the existing runs have already been completed (or even partially completed, in the case of batchTriggerAndWait)

- TaskRunExecutionResult.id is now the run friendlyId, not the attempt friendlyId
- A single TaskRun can now have many batchItems, in the case of batchTriggerAndWait while using idempotency keys
- A run’s idempotencyKey is now added to the ctx as well as the TaskEvent and displayed in the span view
- When resolving batchTriggerAndWait, the runtimes no longer reject promises, leading to an error in the parent task

* Remove the default queue concurrency limit as we now have env and org concurrency limits

* Use the run friendlyId in the completion result id

* Added some error logging

Author: ericallam

.changeset/sharp-emus-compare.md

@@ -0,0 +1,13 @@
+---
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+When using idempotency keys, triggerAndWait and batchTriggerAndWait will still work even if the existing runs have already been completed (or even partially completed, in the case of batchTriggerAndWait)
+
+- TaskRunExecutionResult.id is now the run friendlyId, not the attempt friendlyId
+- A single TaskRun can now have many batchItems, in the case of batchTriggerAndWait while using idempotency keys
+- A run’s idempotencyKey is now added to the ctx as well as the TaskEvent and displayed in the span view
+- When resolving batchTriggerAndWait, the runtimes no longer reject promises, leading to an error in the parent task
+

apps/webapp/app/components/runs/RunFilters.tsx

@@ -32,6 +32,7 @@ import {
 import { TimeFrameFilter } from "./TimeFrameFilter";
 import { Button } from "../primitives/Buttons";
 import { useCallback } from "react";
+import assertNever from "assert-never";
 
 export function RunsFilters() {
   const navigate = useNavigate();
@@ -182,8 +183,7 @@ export function FilterStatusIcon({
     case "FAILED":
       return <XCircleIcon className={cn(filterStatusClassNameColor(status), className)} />;
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -205,8 +205,7 @@ export function filterStatusTitle(status: FilterableStatus): string {
     case "TIMEDOUT":
       return "Timed out";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -228,8 +227,7 @@ export function filterStatusClassNameColor(status: FilterableStatus): string {
     case "TIMEDOUT":
       return "text-amber-300";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }

apps/webapp/app/components/runs/RunStatuses.tsx

@@ -11,6 +11,7 @@ import type { JobRunStatus } from "@trigger.dev/database";
 import { cn } from "~/utils/cn";
 import { Spinner } from "../primitives/Spinner";
 import { z } from "zod";
+import assertNever from "assert-never";
 
 export function RunStatus({ status }: { status: JobRunStatus }) {
   return (
@@ -51,8 +52,7 @@ export function RunStatusIcon({ status, className }: { status: JobRunStatus; cla
     case "CANCELED":
       return <NoSymbolIcon className={cn(runStatusClassNameColor(status), className)} />;
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -89,8 +89,7 @@ export function runStatusTitle(status: JobRunStatus): string {
     case "INVALID_PAYLOAD":
       return "Invalid payload";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -123,8 +122,7 @@ export function runStatusClassNameColor(status: JobRunStatus): string {
     case "CANCELED":
       return "text-charcoal-500";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }

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

@@ -5,6 +5,7 @@ import {
   XCircleIcon,
 } from "@heroicons/react/20/solid";
 import { WorkerDeploymentStatus } from "@trigger.dev/database";
+import assertNever from "assert-never";
 import { Spinner } from "~/components/primitives/Spinner";
 import { cn } from "~/utils/cn";
 
@@ -54,8 +55,7 @@ export function DeploymentStatusIcon({
         />
       );
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -74,8 +74,7 @@ export function deploymentStatusClassNameColor(status: WorkerDeploymentStatus):
     case "FAILED":
       return "text-error";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -97,8 +96,7 @@ export function deploymentStatusTitle(status: WorkerDeploymentStatus): string {
     case "FAILED":
       return "Failed";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }

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

@@ -8,6 +8,7 @@ import {
 } from "@heroicons/react/20/solid";
 import type { TaskRunAttemptStatus as TaskRunAttemptStatusType } from "@trigger.dev/database";
 import { TaskRunAttemptStatus } from "@trigger.dev/database";
+import assertNever from "assert-never";
 import { SnowflakeIcon } from "lucide-react";
 import { Spinner } from "~/components/primitives/Spinner";
 import { cn } from "~/utils/cn";
@@ -72,8 +73,7 @@ export function TaskRunAttemptStatusIcon({
     case "COMPLETED":
       return <CheckCircleIcon className={cn(runAttemptStatusClassNameColor(status), className)} />;
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -99,8 +99,7 @@ export function runAttemptStatusClassNameColor(status: ExtendedTaskAttemptStatus
     case "COMPLETED":
       return "text-success";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -126,8 +125,7 @@ export function runAttemptStatusTitle(status: ExtendedTaskAttemptStatus | null):
     case "COMPLETED":
       return "Completed";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }

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

@@ -10,6 +10,7 @@ import {
   XCircleIcon,
 } from "@heroicons/react/20/solid";
 import { TaskRunStatus } from "@trigger.dev/database";
+import assertNever from "assert-never";
 import { SnowflakeIcon } from "lucide-react";
 import { Spinner } from "~/components/primitives/Spinner";
 import { cn } from "~/utils/cn";
@@ -88,8 +89,7 @@ export function TaskRunStatusIcon({
       return <FireIcon className={cn(runStatusClassNameColor(status), className)} />;
 
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -120,8 +120,7 @@ export function runStatusClassNameColor(status: TaskRunStatus): string {
     case "CRASHED":
       return "text-error";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }
@@ -153,8 +152,7 @@ export function runStatusTitle(status: TaskRunStatus): string {
     case "CRASHED":
       return "Crashed";
     default: {
-      const _exhaustiveCheck: never = status;
-      throw new Error(`Non-exhaustive match for value: ${status}`);
+      assertNever(status);
     }
   }
 }

apps/webapp/app/env.server.ts

@@ -76,7 +76,6 @@ const EnvironmentSchema = z.object({
   REDIS_PASSWORD: z.string().optional(),
   REDIS_TLS_DISABLED: z.string().optional(),
 
-  DEFAULT_QUEUE_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(5),
   DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1),

apps/webapp/app/models/taskRun.server.ts

@@ -0,0 +1,128 @@
+import {
+  TaskRunError,
+  TaskRunExecutionResult,
+  TaskRunFailedExecutionResult,
+  TaskRunSuccessfulExecutionResult,
+} from "@trigger.dev/core/v3";
+import {
+  BatchTaskRunItemStatus,
+  TaskRun,
+  TaskRunAttempt,
+  TaskRunAttemptStatus,
+  TaskRunStatus,
+} from "@trigger.dev/database";
+import { assertNever } from "assert-never";
+import { logger } from "~/services/logger.server";
+
+const SUCCESSFUL_STATUSES = [TaskRunStatus.COMPLETED_SUCCESSFULLY];
+const FAILURE_STATUSES = [
+  TaskRunStatus.CANCELED,
+  TaskRunStatus.INTERRUPTED,
+  TaskRunStatus.COMPLETED_WITH_ERRORS,
+  TaskRunStatus.SYSTEM_FAILURE,
+  TaskRunStatus.CRASHED,
+];
+
+export type TaskRunWithAttempts = TaskRun & {
+  attempts: TaskRunAttempt[];
+};
+
+export function executionResultForTaskRun(
+  taskRun: TaskRunWithAttempts
+): TaskRunExecutionResult | undefined {
+  if (SUCCESSFUL_STATUSES.includes(taskRun.status)) {
+    // find the last attempt that was successful
+    const attempt = taskRun.attempts.find((a) => a.status === TaskRunAttemptStatus.COMPLETED);
+
+    if (!attempt) {
+      logger.error("Task run is successful but no successful attempt found", {
+        taskRunId: taskRun.id,
+        taskRunStatus: taskRun.status,
+        taskRunAttempts: taskRun.attempts.map((a) => a.status),
+      });
+
+      return undefined;
+    }
+
+    return {
+      ok: true,
+      id: taskRun.friendlyId,
+      output: attempt.output ?? undefined,
+      outputType: attempt.outputType,
+    } satisfies TaskRunSuccessfulExecutionResult;
+  }
+
+  if (FAILURE_STATUSES.includes(taskRun.status)) {
+    if (taskRun.status === TaskRunStatus.CANCELED) {
+      return {
+        ok: false,
+        id: taskRun.friendlyId,
+        error: {
+          type: "INTERNAL_ERROR",
+          code: "TASK_RUN_CANCELLED",
+        },
+      } satisfies TaskRunFailedExecutionResult;
+    }
+
+    const attempt = taskRun.attempts.find((a) => a.status === TaskRunAttemptStatus.FAILED);
+
+    if (!attempt) {
+      logger.error("Task run is failed but no failed attempt found", {
+        taskRunId: taskRun.id,
+        taskRunStatus: taskRun.status,
+        taskRunAttempts: taskRun.attempts.map((a) => a.status),
+      });
+
+      return undefined;
+    }
+
+    const error = TaskRunError.safeParse(attempt.error);
+
+    if (!error.success) {
+      logger.error("Failed to parse error from failed task run attempt", {
+        taskRunId: taskRun.id,
+        taskRunStatus: taskRun.status,
+        taskRunAttempts: taskRun.attempts.map((a) => a.status),
+        error: attempt.error,
+      });
+
+      return {
+        ok: false,
+        id: taskRun.friendlyId,
+        error: {
+          type: "INTERNAL_ERROR",
+          code: "CONFIGURED_INCORRECTLY",
+        },
+      } satisfies TaskRunFailedExecutionResult;
+    }
+
+    return {
+      ok: false,
+      id: taskRun.friendlyId,
+      error: error.data,
+    } satisfies TaskRunFailedExecutionResult;
+  }
+}
+
+export function batchTaskRunItemStatusForRunStatus(status: TaskRunStatus): BatchTaskRunItemStatus {
+  switch (status) {
+    case TaskRunStatus.COMPLETED_SUCCESSFULLY:
+      return BatchTaskRunItemStatus.COMPLETED;
+    case TaskRunStatus.CANCELED:
+    case TaskRunStatus.INTERRUPTED:
+    case TaskRunStatus.COMPLETED_WITH_ERRORS:
+    case TaskRunStatus.SYSTEM_FAILURE:
+    case TaskRunStatus.CRASHED:
+    case TaskRunStatus.COMPLETED_WITH_ERRORS:
+      return BatchTaskRunItemStatus.FAILED;
+    case TaskRunStatus.PENDING:
+    case TaskRunStatus.WAITING_FOR_DEPLOY:
+    case TaskRunStatus.WAITING_TO_RESUME:
+    case TaskRunStatus.RETRYING_AFTER_FAILURE:
+    case TaskRunStatus.EXECUTING:
+    case TaskRunStatus.PAUSED:
+      return BatchTaskRunItemStatus.PENDING;
+    default:
+      assertNever(status);
+  }
+}

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

@@ -0,0 +1,46 @@
+import { BatchTaskRunExecutionResult } from "@trigger.dev/core/v3";
+import { executionResultForTaskRun } from "~/models/taskRun.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { BasePresenter } from "./basePresenter.server";
+
+export class ApiBatchResultsPresenter extends BasePresenter {
+  public async call(
+    friendlyId: string,
+    env: AuthenticatedEnvironment
+  ): Promise<BatchTaskRunExecutionResult | undefined> {
+    return this.traceWithEnv("call", env, async (span) => {
+      const batchRun = await this._prisma.batchTaskRun.findUnique({
+        where: {
+          friendlyId,
+          runtimeEnvironmentId: env.id,
+        },
+        include: {
+          items: {
+            include: {
+              taskRun: {
+                include: {
+                  attempts: {
+                    orderBy: {
+                      createdAt: "desc",
+                    },
+                  },
+                },
+              },
+            },
+          },
+        },
+      });
+
+      if (!batchRun) {
+        return undefined;
+      }
+
+      return {
+        id: batchRun.friendlyId,
+        items: batchRun.items
+          .map((item) => executionResultForTaskRun(item.taskRun))
+          .filter(Boolean),
+      };
+    });
+  }
+}