Add ttl option when triggering tasks, expire runs after ttl

Dev runs expire in 10m by default

Author: ericallam

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

@@ -8,6 +8,7 @@ import {
   NoSymbolIcon,
   PauseCircleIcon,
   RectangleStackIcon,
+  TrashIcon,
   XCircleIcon,
 } from "@heroicons/react/20/solid";
 import { TaskRunStatus } from "@trigger.dev/database";
@@ -30,6 +31,7 @@ export const allTaskRunStatuses = [
   "PAUSED",
   "INTERRUPTED",
   "SYSTEM_FAILURE",
+  "EXPIRED",
 ] as const satisfies Readonly<Array<TaskRunStatus>>;
 
 export const filterableTaskRunStatuses = [
@@ -45,6 +47,7 @@ export const filterableTaskRunStatuses = [
   "CRASHED",
   "INTERRUPTED",
   "SYSTEM_FAILURE",
+  "EXPIRED",
 ] as const satisfies Readonly<Array<TaskRunStatus>>;
 
 const taskRunStatusDescriptions: Record<TaskRunStatus, string> = {
@@ -61,6 +64,7 @@ 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", "DELAYED"];
@@ -78,6 +82,7 @@ export const FINISHED_STATUSES: TaskRunStatus[] = [
   "INTERRUPTED",
   "SYSTEM_FAILURE",
   "CRASHED",
+  "EXPIRED",
 ];
 
 export function descriptionForTaskRunStatus(status: TaskRunStatus): string {
@@ -139,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);
@@ -161,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";
@@ -206,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

@@ -119,6 +119,7 @@ export function TaskRunsTable({
           <TableHeaderCell>Test</TableHeaderCell>
           <TableHeaderCell>Created at</TableHeaderCell>
           <TableHeaderCell>Delayed until</TableHeaderCell>
+          <TableHeaderCell>TTL</TableHeaderCell>
           <TableHeaderCell>
             <span className="sr-only">Go to page</span>
           </TableHeaderCell>
@@ -191,6 +192,7 @@ export function TaskRunsTable({
                 <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

@@ -41,6 +41,7 @@ export const TaskRunStatus = {
   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,6 +118,7 @@ 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:

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

@@ -117,6 +117,8 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
         output: $output,
         outputPresignedUrl: $outputPresignedUrl,
         isTest: taskRun.isTest,
+        ttl: taskRun.ttl ?? undefined,
+        expiredAt: taskRun.expiredAt ?? undefined,
         schedule: taskRun.schedule
           ? {
               id: taskRun.schedule.friendlyId,
@@ -209,6 +211,9 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
       case "COMPLETED_WITH_ERRORS": {
         return "FAILED";
       }
+      case "EXPIRED": {
+        return "EXPIRED";
+      }
       default: {
         assertNever(status);
       }

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

@@ -211,6 +211,8 @@ export class ApiRunListPresenter extends BasePresenter {
           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,
@@ -269,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

@@ -163,6 +163,8 @@ export class RunListPresenter extends BasePresenter {
         isTest: boolean;
         spanId: string;
         idempotencyKey: string | null;
+        ttl: string | null;
+        expiredAt: Date | null;
       }[]
     >`
     SELECT
@@ -180,7 +182,9 @@ export class RunListPresenter extends BasePresenter {
     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
@@ -297,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/services/worker.server.ts

@@ -48,6 +48,7 @@ import { RequeueV2Message } from "~/v3/marqs/requeueV2Message.server";
 import { MarqsConcurrencyMonitor } from "~/v3/marqs/concurrencyMonitor.server";
 import { reportUsageEvent } from "~/v3/openMeter.server";
 import { EnqueueDelayedRunService } from "~/v3/services/enqueueDelayedRun.server";
+import { ExpireEnqueuedRunService } from "~/v3/services/expireEnqueuedRun.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -181,6 +182,9 @@ const workerCatalog = {
   "v3.enqueueDelayedRun": z.object({
     runId: z.string(),
   }),
+  "v3.expireRun": z.object({
+    runId: z.string(),
+  }),
 };
 
 const executionWorkerCatalog = {
@@ -682,6 +686,15 @@ function getWorkerQueue() {
         handler: async (payload, job) => {
           const service = new EnqueueDelayedRunService();
 
+          return await service.call(payload.runId);
+        },
+      },
+      "v3.expireRun": {
+        priority: 0,
+        maxAttempts: 8,
+        handler: async (payload, job) => {
+          const service = new ExpireEnqueuedRunService();
+
           return await service.call(payload.runId);
         },
       },

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

@@ -69,6 +69,7 @@ export class RequeueTaskRunService extends BaseService {
       case "CRASHED":
       case "COMPLETED_WITH_ERRORS":
       case "COMPLETED_SUCCESSFULLY":
+      case "EXPIRED":
       case "CANCELED": {
         logger.debug("[RequeueTaskRunService] Task run is completed", { taskRun });
 

apps/webapp/app/v3/services/enqueueDelayedRun.server.ts

@@ -1,6 +1,9 @@
 import { logger } from "~/services/logger.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
+import { parseNaturalLanguageDuration } from "./triggerTask.server";
+import { workerQueue } from "~/services/worker.server";
+import { $transaction } from "~/db.server";
 
 export class EnqueueDelayedRunService extends BaseService {
   public async call(runId: string) {
@@ -34,14 +37,28 @@ export class EnqueueDelayedRunService extends BaseService {
       return;
     }
 
-    await this._prisma.taskRun.update({
-      where: {
-        id: run.id,
-      },
-      data: {
-        status: "PENDING",
-        queuedAt: new Date(),
-      },
+    await $transaction(this._prisma, async (tx) => {
+      await tx.taskRun.update({
+        where: {
+          id: run.id,
+        },
+        data: {
+          status: "PENDING",
+          queuedAt: new Date(),
+        },
+      });
+
+      if (run.ttl) {
+        const expireAt = parseNaturalLanguageDuration(run.ttl);
+
+        if (expireAt) {
+          await workerQueue.enqueue(
+            "v3.expireRun",
+            { runId: run.id },
+            { tx, runAt: expireAt, jobKey: `v3.expireRun.${run.id}` }
+          );
+        }
+      }
     });
 
     await marqs?.enqueueMessage(

apps/webapp/app/v3/services/expireEnqueuedRun.server.ts

@@ -0,0 +1,53 @@
+import { logger } from "~/services/logger.server";
+import { marqs } from "~/v3/marqs/index.server";
+import { BaseService } from "./baseService.server";
+
+export class ExpireEnqueuedRunService extends BaseService {
+  public async call(runId: string) {
+    const run = await this._prisma.taskRun.findUnique({
+      where: {
+        id: runId,
+      },
+      include: {
+        runtimeEnvironment: {
+          include: {
+            organization: true,
+            project: true,
+          },
+        },
+      },
+    });
+
+    if (!run) {
+      logger.debug("Could not find enqueued run to expire", {
+        runId,
+      });
+
+      return;
+    }
+
+    if (run.status !== "PENDING") {
+      logger.debug("Run cannot be expired because it's not in PENDING status", {
+        run,
+      });
+
+      return;
+    }
+
+    logger.debug("Expiring enqueued run", {
+      run,
+    });
+
+    await this._prisma.taskRun.update({
+      where: {
+        id: run.id,
+      },
+      data: {
+        status: "EXPIRED",
+        expiredAt: new Date(),
+      },
+    });
+
+    await marqs?.acknowledgeMessage(run.id);
+  }
+}

apps/webapp/app/v3/services/triggerTask.server.ts

@@ -14,7 +14,6 @@ import { generateFriendlyId } from "../friendlyIdentifiers";
 import { uploadToObjectStore } from "../r2.server";
 import { startActiveSpan } from "../tracer.server";
 import { BaseService } from "./baseService.server";
-import { $transaction } from "~/db.server";
 
 export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;
@@ -39,6 +38,11 @@ export class TriggerTaskService extends BaseService {
       const idempotencyKey = options.idempotencyKey ?? body.options?.idempotencyKey;
       const delayUntil = await parseDelay(body.options?.delay);
 
+      const ttl =
+        typeof body.options?.ttl === "number"
+          ? stringifyDuration(body.options?.ttl)
+          : body.options?.ttl ?? (environment.type === "DEVELOPMENT" ? "10m" : undefined);
+
       const existingRun = idempotencyKey
         ? await this._prisma.taskRun.findUnique({
             where: {
@@ -135,6 +139,7 @@ export class TriggerTaskService extends BaseService {
                   isTest: body.options?.test ?? false,
                   delayUntil,
                   queuedAt: delayUntil ? undefined : new Date(),
+                  ttl,
                 },
               });
 
@@ -229,6 +234,18 @@ export class TriggerTaskService extends BaseService {
                 );
               }
 
+              if (!taskRun.delayUntil && taskRun.ttl) {
+                const expireAt = parseNaturalLanguageDuration(taskRun.ttl);
+
+                if (expireAt) {
+                  await workerQueue.enqueue(
+                    "v3.expireRun",
+                    { runId: taskRun.id },
+                    { tx, runAt: expireAt, jobKey: `v3.expireRun.${taskRun.id}` }
+                  );
+                }
+              }
+
               return taskRun;
             },
             async (_, tx) => {
@@ -341,7 +358,7 @@ export async function parseDelay(value?: string | Date): Promise<Date | undefine
   }
 }
 
-function parseNaturalLanguageDuration(duration: string): Date | undefined {
+export function parseNaturalLanguageDuration(duration: string): Date | undefined {
   const regexPattern = /^(\d+w)?(\d+d)?(\d+h)?(\d+m)?(\d+s)?$/;
 
   const result: Date = new Date();
@@ -392,3 +409,25 @@ function parseNaturalLanguageDuration(duration: string): Date | undefined {
 
   return undefined;
 }
+
+function stringifyDuration(seconds: number): string | undefined {
+  if (seconds <= 0) {
+    return;
+  }
+
+  const units = {
+    w: Math.floor(seconds / 604800),
+    d: Math.floor((seconds % 604800) / 86400),
+    h: Math.floor((seconds % 86400) / 3600),
+    m: Math.floor((seconds % 3600) / 60),
+    s: Math.floor(seconds % 60),
+  };
+
+  // Filter the units having non-zero values and join them
+  const result: string = Object.entries(units)
+    .filter(([unit, val]) => val != 0)
+    .map(([unit, val]) => `${val}${unit}`)
+    .join("");
+
+  return result;
+}

apps/webapp/app/v3/taskStatus.ts

@@ -39,6 +39,7 @@ export const FINAL_RUN_STATUSES: TaskRunStatus[] = [
   "COMPLETED_WITH_ERRORS",
   "INTERRUPTED",
   "SYSTEM_FAILURE",
+  "EXPIRED",
 ];
 export const FINAL_ATTEMPT_STATUSES: TaskRunAttemptStatus[] = ["CANCELED", "COMPLETED", "FAILED"];