v4: simplified release concurrency system and status changes (#2284)

* WIP

* Make release concurrency system extremely simple, everything just releases all the time

* update the deadlock detection to use the new lockedQueueReleaseConcurrencyOnWaitpoint column

* WIP new release concurrency system

* Remove releaseConcurrency and releaseConcurrencyOnWaitpoint

Also removed deadlock detection, and added environment burst concurrency

* 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

* Introduce the new "current dequeued concurrency set"

* Remove QUEUED_EXECUTING because we no longer "eagerly" release before checkpointing

* Remove waitpoint test for QUEUED_EXECUTING

* Add isWaiting

* Add changeset

* Use createdAt for ordering realtime runs instead of number

* Clarify the envCurrentDequeuedKey usage

* mock the db.server file to fix the tests

* Updated changset "EXECUTED" -> "EXECUTING"

---------

Co-authored-by: Matt Aitken <[email protected]>

Author: ericallam

.changeset/little-birds-appear.md

@@ -0,0 +1,45 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Removes the `releaseConcurrencyOnWaitpoint` option on queues and the `releaseConcurrency` option on various wait functions. Replaced with the following default behavior:
+
+- Concurrency is never released when a run is first blocked via a waitpoint, at either the env or queue level.
+- Concurrency is always released when a run is checkpointed and shutdown, at both the env and queue level.
+
+Additionally, environment concurrency limits now have a new "Burst Factor", defaulting to 2.0x. The "Burst Factor" allows the environment-wide concurrency limit to be higher than any individual queue's concurrency limit. For example, if you have an environment concurrency limit of 100, and a Burst Factor of 2.0x, then you can execute up to 200 runs concurrently, but any one task/queue can still only execute 100 runs concurrently.
+
+We've done some work cleaning up the run statuses. The new statuses are:
+
+- `PENDING_VERSION`: Task is waiting for a version update because it cannot execute without additional information (task, queue, etc.)
+- `QUEUED`: Task is waiting to be executed by a worker
+- `DEQUEUED`: Task has been dequeued and is being sent to a worker to start executing.
+- `EXECUTING`: Task is currently being executed by a worker
+- `WAITING`: Task has been paused by the system, and will be resumed by the system
+- `COMPLETED`: Task has been completed successfully
+- `CANCELED`: Task has been canceled by the user
+- `FAILED`: Task has failed to complete, due to an error in the system
+- `CRASHED`: Task has crashed and won't be retried, most likely the worker ran out of resources, e.g. memory or storage
+- `SYSTEM_FAILURE`: Task has failed to complete, due to an error in the system
+- `DELAYED`: Task has been scheduled to run at a specific time
+- `EXPIRED`: Task has expired and won't be executed
+- `TIMED_OUT`: Task has reached it's maxDuration and has been stopped
+
+We've removed the following statuses:
+
+- `WAITING_FOR_DEPLOY`: This is no longer used, and is replaced by `PENDING_VERSION`
+- `FROZEN`: This is no longer used, and is replaced by `WAITING`
+- `INTERRUPTED`: This is no longer used
+- `REATTEMPTING`: This is no longer used, and is replaced by `EXECUTING`
+
+We've also added "boolean" helpers to runs returned via the API and from Realtime:
+
+- `isQueued`: Returns true when the status is `QUEUED`, `PENDING_VERSION`, or `DELAYED`
+- `isExecuting`: Returns true when the status is `EXECUTING`, `DEQUEUED`. These count against your concurrency limits.
+- `isWaiting`: Returns true when the status is `WAITING`. These do not count against your concurrency limits.
+- `isCompleted`: Returns true when the status is any of the completed statuses.
+- `isCanceled`: Returns true when the status is `CANCELED`
+- `isFailed`: Returns true when the status is any of the failed statuses.
+- `isSuccess`: Returns true when the status is `COMPLETED`
+
+This change adds the ability to easily detect which runs are being counted against your concurrency limit by filtering for both `EXECUTING` or `DEQUEUED`.

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/env.server.ts

@@ -200,6 +200,7 @@ const EnvironmentSchema = z.object({
   PUBSUB_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
 
   DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(100),
+  DEFAULT_ENV_EXECUTION_CONCURRENCY_BURST_FACTOR: z.coerce.number().default(1.0),
   DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(300),
   DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1),
 

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: