v3: Delayed runs and run ttl expiration (#1193)

* v3: Trigger delayed runs and reschedule them

* Create a `@trigger.dev/core/v3/schemas` export

* fixed the `@trigger.dev/core/v3/schemas` export

* Small docs tweak

* Add ttl option when triggering tasks, expire runs after ttl

Dev runs expire in 10m by default

Author: ericallam

.changeset/modern-stingrays-end.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+v3: Trigger delayed runs and reschedule them

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

@@ -3,10 +3,12 @@ import {
   BoltSlashIcon,
   BugAntIcon,
   CheckCircleIcon,
+  ClockIcon,
   FireIcon,
   NoSymbolIcon,
   PauseCircleIcon,
   RectangleStackIcon,
+  TrashIcon,
   XCircleIcon,
 } from "@heroicons/react/20/solid";
 import { TaskRunStatus } from "@trigger.dev/database";
@@ -16,6 +18,7 @@ import { Spinner } from "~/components/primitives/Spinner";
 import { cn } from "~/utils/cn";
 
 export const allTaskRunStatuses = [
+  "DELAYED",
   "WAITING_FOR_DEPLOY",
   "PENDING",
   "EXECUTING",
@@ -28,10 +31,12 @@ export const allTaskRunStatuses = [
   "PAUSED",
   "INTERRUPTED",
   "SYSTEM_FAILURE",
+  "EXPIRED",
 ] as const satisfies Readonly<Array<TaskRunStatus>>;
 
 export const filterableTaskRunStatuses = [
   "WAITING_FOR_DEPLOY",
+  "DELAYED",
   "PENDING",
   "EXECUTING",
   "RETRYING_AFTER_FAILURE",
@@ -42,9 +47,11 @@ export const filterableTaskRunStatuses = [
   "CRASHED",
   "INTERRUPTED",
   "SYSTEM_FAILURE",
+  "EXPIRED",
 ] as const satisfies Readonly<Array<TaskRunStatus>>;
 
 const taskRunStatusDescriptions: Record<TaskRunStatus, string> = {
+  DELAYED: "Task has been delayed and is waiting to be executed",
   PENDING: "Task is waiting to be executed",
   WAITING_FOR_DEPLOY: "Task needs to be deployed first to start executing",
   EXECUTING: "Task is currently being executed",
@@ -57,9 +64,10 @@ const taskRunStatusDescriptions: Record<TaskRunStatus, string> = {
   SYSTEM_FAILURE: "Task has failed due to a system failure",
   PAUSED: "Task has been paused by the user",
   CRASHED: "Task has crashed and won't be retried",
+  EXPIRED: "Task has surpassed its ttl and won't be executed",
 };
 
-export const QUEUED_STATUSES: TaskRunStatus[] = ["PENDING", "WAITING_FOR_DEPLOY"];
+export const QUEUED_STATUSES: TaskRunStatus[] = ["PENDING", "WAITING_FOR_DEPLOY", "DELAYED"];
 
 export const RUNNING_STATUSES: TaskRunStatus[] = [
   "EXECUTING",
@@ -74,6 +82,7 @@ export const FINISHED_STATUSES: TaskRunStatus[] = [
   "INTERRUPTED",
   "SYSTEM_FAILURE",
   "CRASHED",
+  "EXPIRED",
 ];
 
 export function descriptionForTaskRunStatus(status: TaskRunStatus): string {
@@ -109,6 +118,8 @@ export function TaskRunStatusIcon({
   className: string;
 }) {
   switch (status) {
+    case "DELAYED":
+      return <ClockIcon className={cn(runStatusClassNameColor(status), className)} />;
     case "PENDING":
       return <RectangleStackIcon className={cn(runStatusClassNameColor(status), className)} />;
     case "WAITING_FOR_DEPLOY":
@@ -133,6 +144,8 @@ export function TaskRunStatusIcon({
       return <BugAntIcon className={cn(runStatusClassNameColor(status), className)} />;
     case "CRASHED":
       return <FireIcon className={cn(runStatusClassNameColor(status), className)} />;
+    case "EXPIRED":
+      return <TrashIcon className={cn(runStatusClassNameColor(status), className)} />;
 
     default: {
       assertNever(status);
@@ -143,6 +156,7 @@ export function TaskRunStatusIcon({
 export function runStatusClassNameColor(status: TaskRunStatus): string {
   switch (status) {
     case "PENDING":
+    case "DELAYED":
       return "text-charcoal-500";
     case "WAITING_FOR_DEPLOY":
       return "text-amber-500";
@@ -154,6 +168,7 @@ export function runStatusClassNameColor(status: TaskRunStatus): string {
     case "PAUSED":
       return "text-amber-300";
     case "CANCELED":
+    case "EXPIRED":
       return "text-charcoal-500";
     case "INTERRUPTED":
       return "text-error";
@@ -173,6 +188,8 @@ export function runStatusClassNameColor(status: TaskRunStatus): string {
 
 export function runStatusTitle(status: TaskRunStatus): string {
   switch (status) {
+    case "DELAYED":
+      return "Delayed";
     case "PENDING":
       return "Queued";
     case "WAITING_FOR_DEPLOY":
@@ -197,6 +214,8 @@ export function runStatusTitle(status: TaskRunStatus): string {
       return "System failure";
     case "CRASHED":
       return "Crashed";
+    case "EXPIRED":
+      return "Expired";
     default: {
       assertNever(status);
     }

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

@@ -118,6 +118,8 @@ export function TaskRunsTable({
           <TableHeaderCell>Duration</TableHeaderCell>
           <TableHeaderCell>Test</TableHeaderCell>
           <TableHeaderCell>Created at</TableHeaderCell>
+          <TableHeaderCell>Delayed until</TableHeaderCell>
+          <TableHeaderCell>TTL</TableHeaderCell>
           <TableHeaderCell>
             <span className="sr-only">Go to page</span>
           </TableHeaderCell>
@@ -187,6 +189,10 @@ export function TaskRunsTable({
                 <TableCell to={path}>
                   {run.createdAt ? <DateTime date={run.createdAt} /> : "–"}
                 </TableCell>
+                <TableCell to={path}>
+                  {run.delayUntil ? <DateTime date={run.delayUntil} /> : "–"}
+                </TableCell>
+                <TableCell to={path}>{run.ttl ?? "–"}</TableCell>
                 <RunActionsCell run={run} path={path} />
               </TableRow>
             );

apps/webapp/app/database-types.ts

@@ -40,6 +40,8 @@ export const TaskRunStatus = {
   COMPLETED_WITH_ERRORS: "COMPLETED_WITH_ERRORS",
   SYSTEM_FAILURE: "SYSTEM_FAILURE",
   CRASHED: "CRASHED",
+  DELAYED: "DELAYED",
+  EXPIRED: "EXPIRED",
 } as const satisfies Record<TaskRunStatusType, TaskRunStatusType>;
 
 export const JobRunStatus = {

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

@@ -118,13 +118,15 @@ export function batchTaskRunItemStatusForRunStatus(
     case TaskRunStatus.COMPLETED_WITH_ERRORS:
     case TaskRunStatus.SYSTEM_FAILURE:
     case TaskRunStatus.CRASHED:
+    case TaskRunStatus.EXPIRED:
       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:
+    case TaskRunStatus.DELAYED:
       return BatchTaskRunItemStatus.PENDING;
     default:
       assertNever(status);

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

@@ -111,11 +111,14 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
         finishedAt: ApiRetrieveRunPresenter.isStatusFinished(apiStatus)
           ? taskRun.updatedAt
           : undefined,
+        delayedUntil: taskRun.delayUntil ?? undefined,
         payload: $payload,
         payloadPresignedUrl: $payloadPresignedUrl,
         output: $output,
         outputPresignedUrl: $outputPresignedUrl,
         isTest: taskRun.isTest,
+        ttl: taskRun.ttl ?? undefined,
+        expiredAt: taskRun.expiredAt ?? undefined,
         schedule: taskRun.schedule
           ? {
               id: taskRun.schedule.friendlyId,
@@ -171,6 +174,9 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
 
   static apiStatusFromRunStatus(status: TaskRunStatus): RunStatus {
     switch (status) {
+      case "DELAYED": {
+        return "DELAYED";
+      }
       case "WAITING_FOR_DEPLOY": {
         return "WAITING_FOR_DEPLOY";
       }
@@ -205,14 +211,17 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
       case "COMPLETED_WITH_ERRORS": {
         return "FAILED";
       }
+      case "EXPIRED": {
+        return "EXPIRED";
+      }
       default: {
         assertNever(status);
       }
     }
   }
 
   static apiBooleanHelpersFromRunStatus(status: RunStatus) {
-    const isQueued = status === "QUEUED" || status === "WAITING_FOR_DEPLOY";
+    const isQueued = status === "QUEUED" || status === "WAITING_FOR_DEPLOY" || status === "DELAYED";
     const isExecuting = status === "EXECUTING" || status === "REATTEMPTING" || status === "FROZEN";
     const isCompleted =
       status === "COMPLETED" ||

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

@@ -209,7 +209,10 @@ export class ApiRunListPresenter extends BasePresenter {
           updatedAt: new Date(run.updatedAt),
           startedAt: run.startedAt ? new Date(run.startedAt) : undefined,
           finishedAt: run.finishedAt ? new Date(run.finishedAt) : undefined,
+          delayedUntil: run.delayUntil ? new Date(run.delayUntil) : undefined,
           isTest: run.isTest,
+          ttl: run.ttl ?? undefined,
+          expiredAt: run.expiredAt ? new Date(run.expiredAt) : undefined,
           env: {
             id: run.environment.id,
             name: run.environment.slug,
@@ -233,6 +236,8 @@ export class ApiRunListPresenter extends BasePresenter {
 
   static apiStatusToRunStatuses(status: RunStatus): TaskRunStatus[] | TaskRunStatus {
     switch (status) {
+      case "DELAYED":
+        return "DELAYED";
       case "WAITING_FOR_DEPLOY": {
         return "WAITING_FOR_DEPLOY";
       }
@@ -266,6 +271,9 @@ export class ApiRunListPresenter extends BasePresenter {
       case "FAILED": {
         return "COMPLETED_WITH_ERRORS";
       }
+      case "EXPIRED": {
+        return "EXPIRED";
+      }
       default: {
         assertNever(status);
       }

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

@@ -158,10 +158,13 @@ export class RunListPresenter extends BasePresenter {
         createdAt: Date;
         startedAt: Date | null;
         lockedAt: Date | null;
+        delayUntil: Date | null;
         updatedAt: Date;
         isTest: boolean;
         spanId: string;
         idempotencyKey: string | null;
+        ttl: string | null;
+        expiredAt: Date | null;
       }[]
     >`
     SELECT
@@ -174,11 +177,14 @@ export class RunListPresenter extends BasePresenter {
     tr.status AS status,
     tr."createdAt" AS "createdAt",
     tr."startedAt" AS "startedAt",
+    tr."delayUntil" AS "delayUntil",
     tr."lockedAt" AS "lockedAt",
     tr."updatedAt" AS "updatedAt",
     tr."isTest" AS "isTest",
     tr."spanId" AS "spanId",
-    tr."idempotencyKey" AS "idempotencyKey"
+    tr."idempotencyKey" AS "idempotencyKey",
+    tr."ttl" AS "ttl",
+    tr."expiredAt" AS "expiredAt"
   FROM
     ${sqlDatabaseSchema}."TaskRun" tr
   LEFT JOIN
@@ -283,6 +289,7 @@ export class RunListPresenter extends BasePresenter {
           createdAt: run.createdAt.toISOString(),
           updatedAt: run.updatedAt.toISOString(),
           startedAt: startedAt ? startedAt.toISOString() : undefined,
+          delayUntil: run.delayUntil ? run.delayUntil.toISOString() : undefined,
           hasFinished,
           finishedAt: hasFinished ? run.updatedAt.toISOString() : undefined,
           isTest: run.isTest,
@@ -294,6 +301,8 @@ export class RunListPresenter extends BasePresenter {
           isCancellable: isCancellableRunStatus(run.status),
           environment: displayableEnvironment(environment, userId),
           idempotencyKey: run.idempotencyKey ? run.idempotencyKey : undefined,
+          ttl: run.ttl ? run.ttl : undefined,
+          expiredAt: run.expiredAt ? run.expiredAt.toISOString() : undefined,
         };
       }),
       pagination: {

apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts

@@ -0,0 +1,85 @@
+import type { ActionFunctionArgs } from "@remix-run/server-runtime";
+import { json } from "@remix-run/server-runtime";
+import { RescheduleRunRequestBody } from "@trigger.dev/core/v3/schemas";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { ApiRetrieveRunPresenter } from "~/presenters/v3/ApiRetrieveRunPresenter.server";
+import { authenticateApiRequest } from "~/services/apiAuth.server";
+import { ServiceValidationError } from "~/v3/services/baseService.server";
+import { RescheduleTaskRunService } from "~/v3/services/rescheduleTaskRun.server";
+
+const ParamsSchema = z.object({
+  runParam: z.string(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // Ensure this is a POST request
+  if (request.method.toUpperCase() !== "POST") {
+    return { status: 405, body: "Method Not Allowed" };
+  }
+
+  // Authenticate the request
+  const authenticationResult = await authenticateApiRequest(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or missing API Key" }, { status: 401 });
+  }
+
+  const parsed = ParamsSchema.safeParse(params);
+
+  if (!parsed.success) {
+    return json({ error: "Invalid or missing run ID" }, { status: 400 });
+  }
+
+  const { runParam } = parsed.data;
+
+  const taskRun = await prisma.taskRun.findUnique({
+    where: {
+      friendlyId: runParam,
+      runtimeEnvironmentId: authenticationResult.environment.id,
+    },
+  });
+
+  if (!taskRun) {
+    return json({ error: "Run not found" }, { status: 404 });
+  }
+
+  const anyBody = await request.json();
+
+  const body = RescheduleRunRequestBody.safeParse(anyBody);
+
+  if (!body.success) {
+    return json({ error: "Invalid request body" }, { status: 400 });
+  }
+
+  const service = new RescheduleTaskRunService();
+
+  try {
+    const updatedRun = await service.call(taskRun, body.data);
+
+    if (!updatedRun) {
+      return json({ error: "An unknown error occurred" }, { status: 500 });
+    }
+
+    const presenter = new ApiRetrieveRunPresenter();
+    const result = await presenter.call(
+      updatedRun.friendlyId,
+      authenticationResult.environment,
+      true
+    );
+
+    if (!result) {
+      return json({ error: "Run not found" }, { status: 404 });
+    }
+
+    return json(result);
+  } catch (error) {
+    if (error instanceof ServiceValidationError) {
+      return json({ error: error.message }, { status: 400 });
+    } else if (error instanceof Error) {
+      return json({ error: error.message }, { status: 500 });
+    } else {
+      return json({ error: "An unknown error occurred" }, { status: 500 });
+    }
+  }
+}

apps/webapp/app/routes/api.v2.runs.$runParam.cancel.ts

@@ -33,6 +33,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
   const taskRun = await prisma.taskRun.findUnique({
     where: {
       friendlyId: runParam,
+      runtimeEnvironmentId: authenticationResult.environment.id,
     },
   });