Added new DEQUEUED status

Cleaned up the API run statuses, including now detecting new clients and not breaking older clients by adding an API version header to all requests

Author: ericallam

apps/webapp/app/api/versions.ts

@@ -0,0 +1,57 @@
+import {
+  API_VERSION_HEADER_NAME,
+  API_VERSION as CORE_API_VERSION,
+} from "@trigger.dev/core/v3/serverOnly";
+import { z } from "zod";
+
+export const CURRENT_API_VERSION = CORE_API_VERSION;
+
+export const NON_SPECIFIC_API_VERSION = "none";
+
+export type API_VERSIONS = typeof CURRENT_API_VERSION | typeof NON_SPECIFIC_API_VERSION;
+
+export function getApiVersion(request: Request): API_VERSIONS {
+  const apiVersion = request.headers.get(API_VERSION_HEADER_NAME);
+
+  if (apiVersion === CURRENT_API_VERSION) {
+    return apiVersion;
+  }
+
+  return NON_SPECIFIC_API_VERSION;
+}
+
+// This has been copied from the core package to allow us to use these types in the webapp
+export const RunStatusUnspecifiedApiVersion = z.enum([
+  /// Task is waiting for a version update because it cannot execute without additional information (task, queue, etc.). Replaces WAITING_FOR_DEPLOY
+  "PENDING_VERSION",
+  /// Task hasn't been deployed yet but is waiting to be executed
+  "WAITING_FOR_DEPLOY",
+  /// Task is waiting to be executed by a worker
+  "QUEUED",
+  /// Task is currently being executed by a worker
+  "EXECUTING",
+  /// Task has failed and is waiting to be retried
+  "REATTEMPTING",
+  /// Task has been paused by the system, and will be resumed by the system
+  "FROZEN",
+  /// Task has been completed successfully
+  "COMPLETED",
+  /// Task has been canceled by the user
+  "CANCELED",
+  /// Task has been completed with errors
+  "FAILED",
+  /// Task has crashed and won't be retried, most likely the worker ran out of resources, e.g. memory or storage
+  "CRASHED",
+  /// Task was interrupted during execution, mostly this happens in development environments
+  "INTERRUPTED",
+  /// Task has failed to complete, due to an error in the system
+  "SYSTEM_FAILURE",
+  /// Task has been scheduled to run at a specific time
+  "DELAYED",
+  /// Task has expired and won't be executed
+  "EXPIRED",
+  /// Task has reached it's maxDuration and has been stopped
+  "TIMED_OUT",
+]);
+
+export type RunStatusUnspecifiedApiVersion = z.infer<typeof RunStatusUnspecifiedApiVersion>;

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

@@ -10,10 +10,11 @@ import {
 import { Form, useFetcher } from "@remix-run/react";
 import { IconToggleLeft } from "@tabler/icons-react";
 import type { BulkActionType, TaskRunStatus, TaskTriggerSource } from "@trigger.dev/database";
-import { ListChecks, ListFilterIcon } from "lucide-react";
+import { ListFilterIcon } from "lucide-react";
 import { matchSorter } from "match-sorter";
 import { type ReactNode, useCallback, useEffect, useMemo, useState } from "react";
 import { z } from "zod";
+import { ListCheckedIcon } from "~/assets/icons/ListCheckedIcon";
 import { StatusIcon } from "~/assets/icons/StatusIcon";
 import { TaskIcon } from "~/assets/icons/TaskIcon";
 import { AppliedFilter } from "~/components/primitives/AppliedFilter";
@@ -55,8 +56,6 @@ import {
   TaskRunStatusCombo,
 } from "./TaskRunStatus";
 import { TaskTriggerSourceIcon } from "./TaskTriggerSource";
-import { ListCheckedIcon } from "~/assets/icons/ListCheckedIcon";
-import { cn } from "~/utils/cn";
 
 export const RunStatus = z.enum(allTaskRunStatuses);
 

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

@@ -24,12 +24,13 @@ export const allTaskRunStatuses = [
   "WAITING_FOR_DEPLOY",
   "PENDING_VERSION",
   "PENDING",
+  "DEQUEUED",
   "EXECUTING",
   "RETRYING_AFTER_FAILURE",
   "WAITING_TO_RESUME",
   "COMPLETED_SUCCESSFULLY",
-  "CANCELED",
   "COMPLETED_WITH_ERRORS",
+  "CANCELED",
   "TIMED_OUT",
   "CRASHED",
   "PAUSED",
@@ -42,16 +43,15 @@ export const filterableTaskRunStatuses = [
   "PENDING_VERSION",
   "DELAYED",
   "PENDING",
-  "WAITING_TO_RESUME",
+  "DEQUEUED",
   "EXECUTING",
-  "RETRYING_AFTER_FAILURE",
+  "WAITING_TO_RESUME",
   "COMPLETED_SUCCESSFULLY",
-  "CANCELED",
   "COMPLETED_WITH_ERRORS",
   "TIMED_OUT",
   "CRASHED",
-  "INTERRUPTED",
   "SYSTEM_FAILURE",
+  "CANCELED",
   "EXPIRED",
 ] as const satisfies Readonly<Array<TaskRunStatus>>;
 
@@ -60,6 +60,7 @@ const taskRunStatusDescriptions: Record<TaskRunStatus, string> = {
   PENDING: "Task is waiting to be executed.",
   PENDING_VERSION: "Run cannot execute until a version includes the task and queue.",
   WAITING_FOR_DEPLOY: "Run cannot execute until a version includes the task and queue.",
+  DEQUEUED: "Task has been dequeued from the queue but is not yet executing.",
   EXECUTING: "Task is currently being executed.",
   RETRYING_AFTER_FAILURE: "Task is being reattempted after a failure.",
   WAITING_TO_RESUME: `You have used a "wait" function. When the wait is complete, the task will resume execution.`,
@@ -82,6 +83,7 @@ export const QUEUED_STATUSES = [
 ] satisfies TaskRunStatus[];
 
 export const RUNNING_STATUSES = [
+  "DEQUEUED",
   "EXECUTING",
   "RETRYING_AFTER_FAILURE",
   "WAITING_TO_RESUME",
@@ -164,6 +166,8 @@ export function TaskRunStatusIcon({
     case "PENDING_VERSION":
     case "WAITING_FOR_DEPLOY":
       return <RectangleStackIcon className={cn(runStatusClassNameColor(status), className)} />;
+    case "DEQUEUED":
+      return <RectangleStackIcon className={cn(runStatusClassNameColor(status), className)} />;
     case "EXECUTING":
       return <Spinner className={cn(runStatusClassNameColor(status), className)} />;
     case "WAITING_TO_RESUME":
@@ -205,6 +209,7 @@ export function runStatusClassNameColor(status: TaskRunStatus): string {
       return "text-amber-500";
     case "EXECUTING":
     case "RETRYING_AFTER_FAILURE":
+    case "DEQUEUED":
       return "text-pending";
     case "WAITING_TO_RESUME":
       return "text-charcoal-500";
@@ -240,6 +245,8 @@ export function runStatusTitle(status: TaskRunStatus): string {
     case "PENDING_VERSION":
     case "WAITING_FOR_DEPLOY":
       return "Pending version";
+    case "DEQUEUED":
+      return "Dequeued";
     case "EXECUTING":
       return "Executing";
     case "WAITING_TO_RESUME":

apps/webapp/app/database-types.ts

@@ -30,6 +30,7 @@ export const TaskRunStatus = {
   PENDING: "PENDING",
   PENDING_VERSION: "PENDING_VERSION",
   WAITING_FOR_DEPLOY: "WAITING_FOR_DEPLOY",
+  DEQUEUED: "DEQUEUED",
   EXECUTING: "EXECUTING",
   WAITING_TO_RESUME: "WAITING_TO_RESUME",
   RETRYING_AFTER_FAILURE: "RETRYING_AFTER_FAILURE",

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

@@ -129,6 +129,7 @@ export function batchTaskRunItemStatusForRunStatus(
     case TaskRunStatus.WAITING_FOR_DEPLOY:
     case TaskRunStatus.WAITING_TO_RESUME:
     case TaskRunStatus.RETRYING_AFTER_FAILURE:
+    case TaskRunStatus.DEQUEUED:
     case TaskRunStatus.EXECUTING:
     case TaskRunStatus.PAUSED:
     case TaskRunStatus.DELAYED:

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

@@ -1,6 +1,5 @@
 import {
   AttemptStatus,
-  RetrieveRunResponse,
   RunStatus,
   SerializedError,
   TaskRunError,
@@ -12,10 +11,12 @@ import {
 } from "@trigger.dev/core/v3";
 import { Prisma, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
 import assertNever from "assert-never";
+import { API_VERSIONS, CURRENT_API_VERSION, RunStatusUnspecifiedApiVersion } from "~/api/versions";
+import { $replica, prisma } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { generatePresignedUrl } from "~/v3/r2.server";
-import { BasePresenter } from "./basePresenter.server";
-import { $replica, prisma } from "~/db.server";
+import { tracer } from "~/v3/tracer.server";
+import { startSpanWithEnv } from "~/v3/tracing.server";
 
 // Build 'select' object
 const commonRunSelect = {
@@ -63,7 +64,9 @@ type CommonRelatedRun = Prisma.Result<
 
 type FoundRun = NonNullable<Awaited<ReturnType<typeof ApiRetrieveRunPresenter.findRun>>>;
 
-export class ApiRetrieveRunPresenter extends BasePresenter {
+export class ApiRetrieveRunPresenter {
+  constructor(private readonly apiVersion: API_VERSIONS) {}
+
   public static async findRun(friendlyId: string, env: AuthenticatedEnvironment) {
     return $replica.taskRun.findFirst({
       where: {
@@ -98,11 +101,8 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
     });
   }
 
-  public async call(
-    taskRun: FoundRun,
-    env: AuthenticatedEnvironment
-  ): Promise<RetrieveRunResponse | undefined> {
-    return this.traceWithEnv("call", env, async (span) => {
+  public async call(taskRun: FoundRun, env: AuthenticatedEnvironment) {
+    return startSpanWithEnv(tracer, "ApiRetrieveRunPresenter.call", env, async () => {
       let $payload: any;
       let $payloadPresignedUrl: string | undefined;
       let $output: any;
@@ -167,7 +167,7 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
       }
 
       return {
-        ...(await createCommonRunStructure(taskRun)),
+        ...(await createCommonRunStructure(taskRun, this.apiVersion)),
         payload: $payload,
         payloadPresignedUrl: $payloadPresignedUrl,
         output: $output,
@@ -180,13 +180,13 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
         attempts: [],
         relatedRuns: {
           root: taskRun.rootTaskRun
-            ? await createCommonRunStructure(taskRun.rootTaskRun)
+            ? await createCommonRunStructure(taskRun.rootTaskRun, this.apiVersion)
             : undefined,
           parent: taskRun.parentTaskRun
-            ? await createCommonRunStructure(taskRun.parentTaskRun)
+            ? await createCommonRunStructure(taskRun.parentTaskRun, this.apiVersion)
             : undefined,
           children: await Promise.all(
-            taskRun.childRuns.map(async (r) => await createCommonRunStructure(r))
+            taskRun.childRuns.map(async (r) => await createCommonRunStructure(r, this.apiVersion))
           ),
         },
       };
@@ -205,7 +205,7 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
     }
   }
 
-  static isStatusFinished(status: RunStatus) {
+  static isStatusFinished(status: RunStatus | RunStatusUnspecifiedApiVersion) {
     return (
       status === "COMPLETED" ||
       status === "FAILED" ||
@@ -216,7 +216,21 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
     );
   }
 
-  static apiStatusFromRunStatus(status: TaskRunStatus): RunStatus {
+  static apiStatusFromRunStatus(
+    status: TaskRunStatus,
+    apiVersion: API_VERSIONS
+  ): RunStatus | RunStatusUnspecifiedApiVersion {
+    switch (apiVersion) {
+      case CURRENT_API_VERSION: {
+        return this.apiStatusFromRunStatusV2(status);
+      }
+      default: {
+        return this.apiStatusFromRunStatusV1(status);
+      }
+    }
+  }
+
+  static apiStatusFromRunStatusV1(status: TaskRunStatus): RunStatusUnspecifiedApiVersion {
     switch (status) {
       case "DELAYED": {
         return "DELAYED";
@@ -237,6 +251,7 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
       case "RETRYING_AFTER_FAILURE": {
         return "REATTEMPTING";
       }
+      case "DEQUEUED":
       case "EXECUTING": {
         return "EXECUTING";
       }
@@ -270,19 +285,77 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
     }
   }
 
-  static apiBooleanHelpersFromTaskRunStatus(status: TaskRunStatus) {
+  static apiStatusFromRunStatusV2(status: TaskRunStatus): RunStatus {
+    switch (status) {
+      case "DELAYED": {
+        return "DELAYED";
+      }
+      case "PENDING_VERSION": {
+        return "PENDING_VERSION";
+      }
+      case "WAITING_FOR_DEPLOY": {
+        return "PENDING_VERSION";
+      }
+      case "PENDING": {
+        return "QUEUED";
+      }
+      case "PAUSED":
+      case "WAITING_TO_RESUME": {
+        return "WAITING";
+      }
+      case "DEQUEUED": {
+        return "DEQUEUED";
+      }
+      case "RETRYING_AFTER_FAILURE":
+      case "EXECUTING": {
+        return "EXECUTING";
+      }
+      case "CANCELED": {
+        return "CANCELED";
+      }
+      case "COMPLETED_SUCCESSFULLY": {
+        return "COMPLETED";
+      }
+      case "SYSTEM_FAILURE": {
+        return "SYSTEM_FAILURE";
+      }
+      case "CRASHED": {
+        return "CRASHED";
+      }
+      case "INTERRUPTED":
+      case "COMPLETED_WITH_ERRORS": {
+        return "FAILED";
+      }
+      case "EXPIRED": {
+        return "EXPIRED";
+      }
+      case "TIMED_OUT": {
+        return "TIMED_OUT";
+      }
+      default: {
+        assertNever(status);
+      }
+    }
+  }
+
+  static apiBooleanHelpersFromTaskRunStatus(status: TaskRunStatus, apiVersion: API_VERSIONS) {
     return ApiRetrieveRunPresenter.apiBooleanHelpersFromRunStatus(
-      ApiRetrieveRunPresenter.apiStatusFromRunStatus(status)
+      ApiRetrieveRunPresenter.apiStatusFromRunStatus(status, apiVersion)
     );
   }
 
-  static apiBooleanHelpersFromRunStatus(status: RunStatus) {
+  static apiBooleanHelpersFromRunStatus(status: RunStatus | RunStatusUnspecifiedApiVersion) {
     const isQueued =
       status === "QUEUED" ||
       status === "WAITING_FOR_DEPLOY" ||
       status === "DELAYED" ||
       status === "PENDING_VERSION";
-    const isExecuting = status === "EXECUTING" || status === "REATTEMPTING" || status === "FROZEN";
+    const isExecuting =
+      status === "EXECUTING" ||
+      status === "REATTEMPTING" ||
+      status === "FROZEN" ||
+      status === "DEQUEUED" ||
+      status === "WAITING";
     const isCompleted =
       status === "COMPLETED" ||
       status === "CANCELED" ||
@@ -358,7 +431,7 @@ async function resolveSchedule(run: CommonRelatedRun) {
   };
 }
 
-async function createCommonRunStructure(run: CommonRelatedRun) {
+async function createCommonRunStructure(run: CommonRelatedRun, apiVersion: API_VERSIONS) {
   const metadata = await parsePacket({
     data: run.metadata ?? undefined,
     dataType: run.metadataType,
@@ -369,7 +442,7 @@ async function createCommonRunStructure(run: CommonRelatedRun) {
     taskIdentifier: run.taskIdentifier,
     idempotencyKey: run.idempotencyKey ?? undefined,
     version: run.lockedToVersion?.version,
-    status: ApiRetrieveRunPresenter.apiStatusFromRunStatus(run.status),
+    status: ApiRetrieveRunPresenter.apiStatusFromRunStatus(run.status, apiVersion),
     createdAt: run.createdAt,
     startedAt: run.startedAt ?? undefined,
     updatedAt: run.updatedAt,
@@ -385,7 +458,7 @@ async function createCommonRunStructure(run: CommonRelatedRun) {
     tags: run.tags
       .map((t: { name: string }) => t.name)
       .sort((a: string, b: string) => a.localeCompare(b)),
-    ...ApiRetrieveRunPresenter.apiBooleanHelpersFromTaskRunStatus(run.status),
+    ...ApiRetrieveRunPresenter.apiBooleanHelpersFromTaskRunStatus(run.status, apiVersion),
     triggerFunction: resolveTriggerFunction(run),
     batchId: run.batch?.friendlyId,
     metadata,