diff --git a/.changeset/modern-stingrays-end.md b/.changeset/modern-stingrays-end.md new file mode 100644 index 000000000..4812eeda3 --- /dev/null +++ b/.changeset/modern-stingrays-end.md @@ -0,0 +1,6 @@ +--- +"@trigger.dev/sdk": patch +"@trigger.dev/core": patch +--- + +v3: Trigger delayed runs and reschedule them diff --git a/apps/webapp/app/components/runs/v3/TaskRunStatus.tsx b/apps/webapp/app/components/runs/v3/TaskRunStatus.tsx index 3e7ef6db6..b35771688 100644 --- a/apps/webapp/app/components/runs/v3/TaskRunStatus.tsx +++ b/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>; 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>; const taskRunStatusDescriptions: Record = { + 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 = { 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 ; case "PENDING": return ; case "WAITING_FOR_DEPLOY": @@ -133,6 +144,8 @@ export function TaskRunStatusIcon({ return ; case "CRASHED": return ; + case "EXPIRED": + return ; 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); } diff --git a/apps/webapp/app/components/runs/v3/TaskRunsTable.tsx b/apps/webapp/app/components/runs/v3/TaskRunsTable.tsx index 2a6ac4fe9..b72b2c076 100644 --- a/apps/webapp/app/components/runs/v3/TaskRunsTable.tsx +++ b/apps/webapp/app/components/runs/v3/TaskRunsTable.tsx @@ -118,6 +118,8 @@ export function TaskRunsTable({ Duration Test Created at + Delayed until + TTL Go to page @@ -187,6 +189,10 @@ export function TaskRunsTable({ {run.createdAt ? : "–"} + + {run.delayUntil ? : "–"} + + {run.ttl ?? "–"} ); diff --git a/apps/webapp/app/database-types.ts b/apps/webapp/app/database-types.ts index 9afc5a5d4..7ea0766ca 100644 --- a/apps/webapp/app/database-types.ts +++ b/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; export const JobRunStatus = { diff --git a/apps/webapp/app/models/taskRun.server.ts b/apps/webapp/app/models/taskRun.server.ts index 7922c1aed..0c78b25ba 100644 --- a/apps/webapp/app/models/taskRun.server.ts +++ b/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: @@ -125,6 +126,7 @@ export function batchTaskRunItemStatusForRunStatus( case TaskRunStatus.RETRYING_AFTER_FAILURE: case TaskRunStatus.EXECUTING: case TaskRunStatus.PAUSED: + case TaskRunStatus.DELAYED: return BatchTaskRunItemStatus.PENDING; default: assertNever(status); diff --git a/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts index 1129810c6..2e09f7977 100644 --- a/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts +++ b/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,6 +211,9 @@ export class ApiRetrieveRunPresenter extends BasePresenter { case "COMPLETED_WITH_ERRORS": { return "FAILED"; } + case "EXPIRED": { + return "EXPIRED"; + } default: { assertNever(status); } @@ -212,7 +221,7 @@ export class ApiRetrieveRunPresenter extends BasePresenter { } 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" || diff --git a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts index 933f58218..2036409e3 100644 --- a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts +++ b/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); } diff --git a/apps/webapp/app/presenters/v3/RunListPresenter.server.ts b/apps/webapp/app/presenters/v3/RunListPresenter.server.ts index e041f2a4b..64b7c46c4 100644 --- a/apps/webapp/app/presenters/v3/RunListPresenter.server.ts +++ b/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: { diff --git a/apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts b/apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts new file mode 100644 index 000000000..117667d01 --- /dev/null +++ b/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 }); + } + } +} diff --git a/apps/webapp/app/routes/api.v2.runs.$runParam.cancel.ts b/apps/webapp/app/routes/api.v2.runs.$runParam.cancel.ts index 8862c8875..1a32a8ce3 100644 --- a/apps/webapp/app/routes/api.v2.runs.$runParam.cancel.ts +++ b/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, }, }); diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts index 034a704ad..7b8bac1e3 100644 --- a/apps/webapp/app/services/worker.server.ts +++ b/apps/webapp/app/services/worker.server.ts @@ -47,6 +47,8 @@ import { ResumeTaskService } from "./tasks/resumeTask.server"; 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({ @@ -177,6 +179,12 @@ const workerCatalog = { }), additionalData: z.record(z.any()).optional(), }), + "v3.enqueueDelayedRun": z.object({ + runId: z.string(), + }), + "v3.expireRun": z.object({ + runId: z.string(), + }), }; const executionWorkerCatalog = { @@ -672,6 +680,24 @@ function getWorkerQueue() { }); }, }, + "v3.enqueueDelayedRun": { + priority: 0, + maxAttempts: 8, + 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); + }, + }, }, }); } diff --git a/apps/webapp/app/v3/requeueTaskRun.server.ts b/apps/webapp/app/v3/requeueTaskRun.server.ts index e2b904998..0ca8497e0 100644 --- a/apps/webapp/app/v3/requeueTaskRun.server.ts +++ b/apps/webapp/app/v3/requeueTaskRun.server.ts @@ -48,6 +48,7 @@ export class RequeueTaskRunService extends BaseService { break; } + case "DELAYED": case "WAITING_FOR_DEPLOY": { logger.debug("[RequeueTaskRunService] Removing task run from queue", { taskRun }); @@ -68,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 }); diff --git a/apps/webapp/app/v3/services/enqueueDelayedRun.server.ts b/apps/webapp/app/v3/services/enqueueDelayedRun.server.ts new file mode 100644 index 000000000..5ae916a54 --- /dev/null +++ b/apps/webapp/app/v3/services/enqueueDelayedRun.server.ts @@ -0,0 +1,72 @@ +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) { + const run = await this._prisma.taskRun.findUnique({ + where: { + id: runId, + }, + include: { + runtimeEnvironment: { + include: { + organization: true, + project: true, + }, + }, + }, + }); + + if (!run) { + logger.debug("Could not find delayed run to enqueue", { + runId, + }); + + return; + } + + if (run.status !== "DELAYED") { + logger.debug("Delayed run cannot be enqueued because it's not in DELAYED status", { + run, + }); + + return; + } + + 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( + run.runtimeEnvironment, + run.queue, + run.id, + { type: "EXECUTE", taskIdentifier: run.taskIdentifier }, + run.concurrencyKey ?? undefined + ); + } +} diff --git a/apps/webapp/app/v3/services/expireEnqueuedRun.server.ts b/apps/webapp/app/v3/services/expireEnqueuedRun.server.ts new file mode 100644 index 000000000..b505afee5 --- /dev/null +++ b/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); + } +} diff --git a/apps/webapp/app/v3/services/rescheduleTaskRun.server.ts b/apps/webapp/app/v3/services/rescheduleTaskRun.server.ts new file mode 100644 index 000000000..4d9461d06 --- /dev/null +++ b/apps/webapp/app/v3/services/rescheduleTaskRun.server.ts @@ -0,0 +1,39 @@ +import { TaskRun } from "@trigger.dev/database"; +import { BaseService, ServiceValidationError } from "./baseService.server"; +import { RescheduleRunRequestBody } from "@trigger.dev/core/v3"; +import { parseDelay } from "./triggerTask.server"; +import { $transaction } from "~/db.server"; +import { workerQueue } from "~/services/worker.server"; + +export class RescheduleTaskRunService extends BaseService { + public async call(taskRun: TaskRun, body: RescheduleRunRequestBody) { + if (taskRun.status !== "DELAYED") { + throw new ServiceValidationError("Cannot reschedule a run that is not delayed"); + } + + const delay = await parseDelay(body.delay); + + if (!delay) { + throw new ServiceValidationError(`Invalid delay: ${body.delay}`); + } + + return await $transaction(this._prisma, async (tx) => { + const updatedRun = await tx.taskRun.update({ + where: { + id: taskRun.id, + }, + data: { + delayUntil: delay, + }, + }); + + await workerQueue.enqueue( + "v3.enqueueDelayedRun", + { runId: taskRun.id }, + { tx, runAt: delay, jobKey: `v3.enqueueDelayedRun.${taskRun.id}` } + ); + + return updatedRun; + }); + } +} diff --git a/apps/webapp/app/v3/services/triggerTask.server.ts b/apps/webapp/app/v3/services/triggerTask.server.ts index cce2cccfd..1648bdef5 100644 --- a/apps/webapp/app/v3/services/triggerTask.server.ts +++ b/apps/webapp/app/v3/services/triggerTask.server.ts @@ -4,15 +4,16 @@ import { TriggerTaskRequestBody, packetRequiresOffloading, } from "@trigger.dev/core/v3"; +import { env } from "~/env.server"; import { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { autoIncrementCounter } from "~/services/autoIncrementCounter.server"; +import { workerQueue } from "~/services/worker.server"; import { marqs, sanitizeQueueName } from "~/v3/marqs/index.server"; import { eventRepository } from "../eventRepository.server"; import { generateFriendlyId } from "../friendlyIdentifiers"; import { uploadToObjectStore } from "../r2.server"; import { startActiveSpan } from "../tracer.server"; import { BaseService } from "./baseService.server"; -import { env } from "~/env.server"; export type TriggerTaskServiceOptions = { idempotencyKey?: string; @@ -35,6 +36,12 @@ export class TriggerTaskService extends BaseService { span.setAttribute("taskId", taskId); 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({ @@ -49,9 +56,19 @@ export class TriggerTaskService extends BaseService { if (existingRun && existingRun.taskIdentifier === taskId) { span.setAttribute("runId", existingRun.friendlyId); + return existingRun; } + const runFriendlyId = generateFriendlyId("run"); + + const payloadPacket = await this.#handlePayloadPacket( + body.payload, + body.options?.payloadType ?? "application/json", + runFriendlyId, + environment + ); + return await eventRepository.traceEvent( taskId, { @@ -76,15 +93,6 @@ export class TriggerTaskService extends BaseService { immediate: true, }, async (event, traceContext) => { - const runFriendlyId = generateFriendlyId("run"); - - const payloadPacket = await this.#handlePayloadPacket( - body.payload, - body.options?.payloadType ?? "application/json", - runFriendlyId, - environment - ); - const run = await autoIncrementCounter.incrementInTransaction( `v3-run:${environment.id}:${taskId}`, async (num, tx) => { @@ -112,7 +120,7 @@ export class TriggerTaskService extends BaseService { const taskRun = await tx.taskRun.create({ data: { - status: "PENDING", + status: delayUntil ? "DELAYED" : "PENDING", number: num, friendlyId: runFriendlyId, runtimeEnvironmentId: environment.id, @@ -129,6 +137,9 @@ export class TriggerTaskService extends BaseService { concurrencyKey: body.options?.concurrencyKey, queue: queueName, isTest: body.options?.test ?? false, + delayUntil, + queuedAt: delayUntil ? undefined : new Date(), + ttl, }, }); @@ -215,6 +226,26 @@ export class TriggerTaskService extends BaseService { } } + if (taskRun.delayUntil) { + await workerQueue.enqueue( + "v3.enqueueDelayedRun", + { runId: taskRun.id }, + { tx, runAt: delayUntil, jobKey: `v3.enqueueDelayedRun.${taskRun.id}` } + ); + } + + 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) => { @@ -238,13 +269,15 @@ export class TriggerTaskService extends BaseService { } // We need to enqueue the task run into the appropriate queue. This is done after the tx completes to prevent a race condition where the task run hasn't been created yet by the time we dequeue. - await marqs?.enqueueMessage( - environment, - run.queue, - run.id, - { type: "EXECUTE", taskIdentifier: taskId }, - body.options?.concurrencyKey - ); + if (run.status === "PENDING") { + await marqs?.enqueueMessage( + environment, + run.queue, + run.id, + { type: "EXECUTE", taskIdentifier: taskId }, + body.options?.concurrencyKey + ); + } return run; } @@ -297,3 +330,104 @@ export class TriggerTaskService extends BaseService { return { dataType: payloadType }; } } + +export async function parseDelay(value?: string | Date): Promise { + if (!value) { + return; + } + + if (value instanceof Date) { + return value; + } + + try { + const date = new Date(value); + + // Check if the date is valid + if (isNaN(date.getTime())) { + return parseNaturalLanguageDuration(value); + } + + if (date.getTime() <= Date.now()) { + return; + } + + return date; + } catch (error) { + return parseNaturalLanguageDuration(value); + } +} + +export function parseNaturalLanguageDuration(duration: string): Date | undefined { + const regexPattern = /^(\d+w)?(\d+d)?(\d+h)?(\d+m)?(\d+s)?$/; + + const result: Date = new Date(); + let hasMatch = false; + + const elements = duration.match(regexPattern); + if (elements) { + if (elements[1]) { + const weeks = Number(elements[1].slice(0, -1)); + if (weeks >= 0) { + result.setDate(result.getDate() + 7 * weeks); + hasMatch = true; + } + } + if (elements[2]) { + const days = Number(elements[2].slice(0, -1)); + if (days >= 0) { + result.setDate(result.getDate() + days); + hasMatch = true; + } + } + if (elements[3]) { + const hours = Number(elements[3].slice(0, -1)); + if (hours >= 0) { + result.setHours(result.getHours() + hours); + hasMatch = true; + } + } + if (elements[4]) { + const minutes = Number(elements[4].slice(0, -1)); + if (minutes >= 0) { + result.setMinutes(result.getMinutes() + minutes); + hasMatch = true; + } + } + if (elements[5]) { + const seconds = Number(elements[5].slice(0, -1)); + if (seconds >= 0) { + result.setSeconds(result.getSeconds() + seconds); + hasMatch = true; + } + } + } + + if (hasMatch) { + return result; + } + + 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; +} diff --git a/apps/webapp/app/v3/taskStatus.ts b/apps/webapp/app/v3/taskStatus.ts index 35c9451a7..13e3c0a6e 100644 --- a/apps/webapp/app/v3/taskStatus.ts +++ b/apps/webapp/app/v3/taskStatus.ts @@ -1,6 +1,7 @@ import type { TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database"; export const CANCELLABLE_RUN_STATUSES: TaskRunStatus[] = [ + "DELAYED", "PENDING", "WAITING_FOR_DEPLOY", "EXECUTING", @@ -38,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"]; diff --git a/docs/images/v3/delayed-runs.png b/docs/images/v3/delayed-runs.png new file mode 100644 index 000000000..e9cd63a11 Binary files /dev/null and b/docs/images/v3/delayed-runs.png differ diff --git a/docs/images/v3/expired-runs.png b/docs/images/v3/expired-runs.png new file mode 100644 index 000000000..7d6177bd7 Binary files /dev/null and b/docs/images/v3/expired-runs.png differ diff --git a/docs/mint.json b/docs/mint.json index 06efa43bb..7daf6d55b 100644 --- a/docs/mint.json +++ b/docs/mint.json @@ -1,8 +1,14 @@ { "$schema": "https://mintlify.com/schema.json", "name": "Trigger.dev", - "openapi": ["/openapi.yml", "/v3-openapi.yaml"], - "versions": ["v3", "v2"], + "openapi": [ + "/openapi.yml", + "/v3-openapi.yaml" + ], + "versions": [ + "v3", + "v2" + ], "api": { "playground": { "mode": "simple" @@ -29,7 +35,6 @@ } }, "theme": "quill", - "modeToggle": { "default": "dark", "isHidden": true @@ -49,7 +54,6 @@ "url": "https://discord.gg/kA47vcd8P6" } ], - "redirects": [ { "source": "/documentation/quickstart", @@ -72,7 +76,9 @@ { "group": "", "version": "v3", - "pages": ["v3/introduction"] + "pages": [ + "v3/introduction" + ] }, { "group": "Getting Started", @@ -95,7 +101,10 @@ "v3/apikeys", { "group": "Task types", - "pages": ["v3/tasks-regular", "v3/tasks-scheduled"] + "pages": [ + "v3/tasks-regular", + "v3/tasks-scheduled" + ] }, "v3/trigger-config" ] @@ -103,7 +112,10 @@ { "group": "Development", "version": "v3", - "pages": ["v3/cli-dev", "v3/run-tests"] + "pages": [ + "v3/cli-dev", + "v3/run-tests" + ] }, { "group": "Deployment", @@ -114,7 +126,9 @@ "v3/github-actions", { "group": "Deployment integrations", - "pages": ["v3/vercel-integration"] + "pages": [ + "v3/vercel-integration" + ] } ] }, @@ -150,7 +164,10 @@ "v3/management/overview", { "group": "Tasks API", - "pages": ["v3/management/tasks/trigger", "v3/management/tasks/batch-trigger"] + "pages": [ + "v3/management/tasks/trigger", + "v3/management/tasks/batch-trigger" + ] }, { "group": "Runs API", @@ -158,7 +175,8 @@ "v3/management/runs/list", "v3/management/runs/retrieve", "v3/management/runs/replay", - "v3/management/runs/cancel" + "v3/management/runs/cancel", + "v3/management/runs/reschedule" ] }, { @@ -187,14 +205,20 @@ }, { "group": "Projects API", - "pages": ["v3/management/projects/runs"] + "pages": [ + "v3/management/projects/runs" + ] } ] }, { "group": "Open source", "version": "v3", - "pages": ["v3/github-repo", "v3/open-source-self-hosting", "v3/open-source-contributing"] + "pages": [ + "v3/github-repo", + "v3/open-source-self-hosting", + "v3/open-source-contributing" + ] }, { "group": "Troubleshooting", @@ -210,7 +234,11 @@ { "group": "Help", "version": "v3", - "pages": ["v3/community", "v3/help-slack", "v3/help-email"] + "pages": [ + "v3/community", + "v3/help-slack", + "v3/help-email" + ] }, { "group": "Getting Started", @@ -402,7 +430,10 @@ "pages": [ { "group": "Airtable", - "pages": ["integrations/apis/airtable", "integrations/apis/airtable-tasks"] + "pages": [ + "integrations/apis/airtable", + "integrations/apis/airtable-tasks" + ] }, { "group": "GitHub", @@ -428,16 +459,25 @@ }, { "group": "Plain", - "pages": ["integrations/apis/plain", "integrations/apis/plain-tasks"] + "pages": [ + "integrations/apis/plain", + "integrations/apis/plain-tasks" + ] }, "integrations/apis/replicate", { "group": "SendGrid", - "pages": ["integrations/apis/sendgrid", "integrations/apis/sendgrid-tasks"] + "pages": [ + "integrations/apis/sendgrid", + "integrations/apis/sendgrid-tasks" + ] }, { "group": "Resend", - "pages": ["integrations/apis/resend", "integrations/apis/resend-tasks"] + "pages": [ + "integrations/apis/resend", + "integrations/apis/resend-tasks" + ] }, { "group": "Shopify", @@ -449,7 +489,10 @@ }, { "group": "Slack", - "pages": ["integrations/apis/slack", "integrations/apis/slack-tasks"] + "pages": [ + "integrations/apis/slack", + "integrations/apis/slack-tasks" + ] }, "integrations/apis/stripe", { @@ -475,7 +518,9 @@ "sdk/triggerclient/constructor", { "group": "Instance properties", - "pages": ["sdk/triggerclient/store"] + "pages": [ + "sdk/triggerclient/store" + ] }, { "group": "Instance methods", @@ -538,7 +583,10 @@ "sdk/dynamictrigger/constructor", { "group": "Instance methods", - "pages": ["sdk/dynamictrigger/register", "sdk/dynamictrigger/unregister"] + "pages": [ + "sdk/dynamictrigger/register", + "sdk/dynamictrigger/unregister" + ] } ] }, @@ -549,7 +597,10 @@ "sdk/dynamicschedule/constructor", { "group": "Instance methods", - "pages": ["sdk/dynamicschedule/register", "sdk/dynamicschedule/unregister"] + "pages": [ + "sdk/dynamicschedule/register", + "sdk/dynamicschedule/unregister" + ] } ] }, @@ -562,7 +613,9 @@ { "group": "HTTP Reference", "version": "v2", - "pages": ["sdk/api-reference/events/create-an-event"] + "pages": [ + "sdk/api-reference/events/create-an-event" + ] }, { "group": "React SDK", @@ -578,7 +631,9 @@ { "group": "Overview", "version": "v2", - "pages": ["examples/introduction"] + "pages": [ + "examples/introduction" + ] } ], "footerSocials": { @@ -586,4 +641,4 @@ "github": "https://github.com/triggerdotdev", "linkedin": "https://www.linkedin.com/company/triggerdotdev" } -} +} \ No newline at end of file diff --git a/docs/v3-openapi.yaml b/docs/v3-openapi.yaml index a6aa84e30..5633ab940 100644 --- a/docs/v3-openapi.yaml +++ b/docs/v3-openapi.yaml @@ -269,6 +269,7 @@ paths: "/api/v1/timezones": get: + security: [] operationId: get_timezones_v1 summary: Get all supported timezones description: Get all supported timezones that schedule tasks support. @@ -424,6 +425,71 @@ paths: await runs.cancel("run_1234"); + "/api/v1/runs/{runId}/reschedule": + parameters: + - $ref: "#/components/parameters/runId" + post: + operationId: reschedule_run_v1 + summary: Rescheduled a delayed run + description: Updates a delayed run with a new delay. Only valid when the run is in the DELAYED state. + requestBody: + required: true + content: + application/json: + schema: + "$ref": "#/components/schemas/RescheduleRunRequestBody" + responses: + "200": + description: Successful request + content: + application/json: + schema: + "$ref": "#/components/schemas/RetrieveRunResponse" + "400": + description: Invalid request + content: + application/json: + schema: + type: object + properties: + error: + type: string + enum: + - Invalid or missing run ID + - Failed to create new run + "401": + description: Unauthorized request + content: + application/json: + schema: + type: object + properties: + error: + type: string + enum: + - Invalid or Missing API key + "404": + description: Resource not found + content: + application/json: + schema: + type: object + properties: + error: + type: string + enum: + - Run not found + tags: + - runs + security: + - secretKey: [] + x-codeSamples: + - lang: typescript + source: |- + import { runs } from "@trigger.dev/sdk/v3"; + + const handle = await runs.reschedule("run_1234", { delay: new Date("2024-06-29T20:45:56.340Z") }); + "/api/v3/runs/{runId}": parameters: - $ref: "#/components/parameters/runId" @@ -1681,6 +1747,18 @@ components: type: string example: slack_123456 required: ["name", "value"] + RescheduleRunRequestBody: + type: object + properties: + delay: + oneOf: + - type: string + description: The duration to delay the run by. The duration should be in the format of `1d`, `6h`, `10m`, `11s`, etc. + example: 1hr + - type: string + format: date-time + description: The Date to delay the run until, e.g. `new Date()` or `"2024-06-25T15:45:26Z"` + example: 2024-06-25T15:45:26Z RetrieveRunResponse: type: object required: @@ -1699,6 +1777,7 @@ components: type: string description: The status of the run enum: + - DELAYED - WAITING_FOR_DEPLOY - QUEUED - EXECUTING @@ -1756,6 +1835,10 @@ components: type: string format: date-time description: The time the run finished + delayedUntil: + type: string + format: date-time + description: If the run was triggered with a delay, this will be the time the run will be enqueued to execute schedule: type: object description: The schedule that triggered the run. Will be omitted if the run was not triggered by a schedule diff --git a/docs/v3/management/runs/reschedule.mdx b/docs/v3/management/runs/reschedule.mdx new file mode 100644 index 000000000..25e59ed96 --- /dev/null +++ b/docs/v3/management/runs/reschedule.mdx @@ -0,0 +1,4 @@ +--- +title: "Reschedule run" +openapi: "v3-openapi POST /api/v1/runs/{runId}/reschedule" +--- diff --git a/docs/v3/triggering.mdx b/docs/v3/triggering.mdx index 2ca7fb0c0..bc02b561a 100644 --- a/docs/v3/triggering.mdx +++ b/docs/v3/triggering.mdx @@ -619,3 +619,94 @@ export const myTask = task({ ### Batch Triggering When using `batchTrigger` or `batchTriggerAndWait`, the total size of all payloads cannot exceed 10MB. This means if you are doing a batch of 100 runs, each payload should be less than 100KB. + +## Delayed runs + +When you want to trigger a task now, but have it run at a later time, you can use the `delay` option: + +```ts +// Delay the task run by 1 hour +await myTask.trigger({ some: "data" }, { delay: "1h" }); +// Delay the task run by 88 seconds +await myTask.trigger({ some: "data" }, { delay: "88s" }); +// Delay the task run by 1 hour and 52 minutes and 18 seconds +await myTask.trigger({ some: "data" }, { delay: "1h52m18s" }); +// Delay until a specific time +await myTask.trigger({ some: "data" }, { delay: "2024-12-01T00:00:00" }); +// Delay using a Date object +await myTask.trigger({ some: "data" }, { delay: new Date(Date.now() + 1000 * 60 * 60) }); +``` + +Runs that are delayed and have not been enqueued yet will display in the dashboard with a "Delayed" status: + +![Delayed run in the dashboard](/images/v3/delayed-runs.png) + + + Delayed runs will be enqueued at the time specified, and will run as soon as possible after that + time, just as a normally triggered run would. + + +You can cancel a delayed run using the `runs.cancel` SDK function: + +```ts +import { runs } from "@trigger.dev/sdk/v3"; + +await runs.cancel("run_1234"); +``` + +You can also reschedule a delayed run using the `runs.reschedule` SDK function: + +```ts +import { runs } from "@trigger.dev/sdk/v3"; + +// The delay option here takes the same format as the trigger delay option +await runs.reschedule("run_1234", { delay: "1h" }); +``` + +The `delay` option is also available when using `batchTrigger`: + +```ts +await myTask.batchTrigger([{ payload: { some: "data" }, options: { delay: "1h" } }]); +``` + +## TTL + +You can set a TTL (time to live) when triggering a task, which will automatically expire the run if it hasn't started within the specified time. This is useful for ensuring that a run doesn't get stuck in the queue for too long. + + + All runs in development have a default `ttl` of 10 minutes. You can disable this by setting the + `ttl` option. + + +```ts +import { myTask } from "./trigger/myTasks"; + +// Expire the run if it hasn't started within 1 hour +await myTask.trigger({ some: "data" }, { ttl: "1h" }); + +// If you specify a number, it will be treated as seconds +await myTask.trigger({ some: "data" }, { ttl: 3600 }); // 1 hour +``` + +When a run is expired, it will be marked as "Expired" in the dashboard: + +![Expired runs in the dashboard](/images/v3/expired-runs.png) + +### Delayed runs and TTL + +When you use both `delay` and `ttl`, the TTL will start counting down from the time the run is enqueued, not from the time the run is triggered. + +So for example, when using the following code: + +```ts +await myTask.trigger({ some: "data" }, { delay: "10m", ttl: "1h" }); +``` + +The timeline would look like this: + +1. The run is created at 12:00:00 +2. The run is enqueued at 12:10:00 +3. The TTL starts counting down from 12:10:00 +4. If the run hasn't started by 13:10:00, it will be expired + +For this reason, the `ttl` option only accepts durations and not absolute timestamps. diff --git a/packages/core/package.json b/packages/core/package.json index a4b973b09..9bf95d06d 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -117,6 +117,14 @@ "require": "./dist/v3/workers/index.js", "types": "./dist/v3/workers/index.d.ts" }, + "./v3/schemas": { + "import": { + "types": "./dist/v3/schemas/index.d.mts", + "default": "./dist/v3/schemas/index.mjs" + }, + "require": "./dist/v3/schemas/index.js", + "types": "./dist/v3/schemas/index.d.ts" + }, "./package.json": "./package.json" }, "typesVersions": { diff --git a/packages/core/src/v3/apiClient/index.ts b/packages/core/src/v3/apiClient/index.ts index 2cb5a08ab..58c9dbd4d 100644 --- a/packages/core/src/v3/apiClient/index.ts +++ b/packages/core/src/v3/apiClient/index.ts @@ -15,6 +15,7 @@ import { ListRunResponseItem, ListScheduleOptions, ReplayRunResponse, + RescheduleRunRequestBody, RetrieveRunResponse, ScheduleObject, TaskRunExecutionResult, @@ -247,6 +248,19 @@ export class ApiClient { ); } + rescheduleRun(runId: string, body: RescheduleRunRequestBody) { + return zodfetch( + RetrieveRunResponse, + `${this.baseUrl}/api/v1/runs/${runId}/reschedule`, + { + method: "POST", + headers: this.#getHeaders(false), + body: JSON.stringify(body), + }, + zodFetchOptions + ); + } + createSchedule(options: CreateScheduleOptions) { return zodfetch(ScheduleObject, `${this.baseUrl}/api/v1/schedules`, { method: "POST", diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts index 1f8291d82..d6666089b 100644 --- a/packages/core/src/v3/schemas/api.ts +++ b/packages/core/src/v3/schemas/api.ts @@ -68,6 +68,8 @@ export const TriggerTaskRequestBody = z.object({ idempotencyKey: z.string().optional(), test: z.boolean().optional(), payloadType: z.string().optional(), + delay: z.string().or(z.coerce.date()).optional(), + ttl: z.string().or(z.number().nonnegative().int()).optional(), }) .optional(), }); @@ -107,6 +109,12 @@ export const GetBatchResponseBody = z.object({ export type GetBatchResponseBody = z.infer; +export const RescheduleRunRequestBody = z.object({ + delay: z.string().or(z.coerce.date()), +}); + +export type RescheduleRunRequestBody = z.infer; + export const GetEnvironmentVariablesResponseBody = z.object({ variables: z.record(z.string()), }); @@ -377,6 +385,10 @@ export const RunStatus = z.enum([ "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", ]); export type RunStatus = z.infer; @@ -426,6 +438,9 @@ const CommonRunFields = { updatedAt: z.coerce.date(), startedAt: z.coerce.date().optional(), finishedAt: z.coerce.date().optional(), + delayedUntil: z.coerce.date().optional(), + ttl: z.string().optional(), + expiredAt: z.coerce.date().optional(), }; export const RetrieveRunResponse = z.object({ diff --git a/packages/core/tsup.config.ts b/packages/core/tsup.config.ts index 420d67820..830fe36a2 100644 --- a/packages/core/tsup.config.ts +++ b/packages/core/tsup.config.ts @@ -17,5 +17,6 @@ export default defineConfig({ "./src/v3/prod/index.ts", "./src/v3/workers/index.ts", "./src/v3/zodfetch.ts", + "./src/v3/schemas/index.ts", ], }); diff --git a/packages/database/prisma/migrations/20240629082837_add_task_run_delay_changes/migration.sql b/packages/database/prisma/migrations/20240629082837_add_task_run_delay_changes/migration.sql new file mode 100644 index 000000000..9fc809f8c --- /dev/null +++ b/packages/database/prisma/migrations/20240629082837_add_task_run_delay_changes/migration.sql @@ -0,0 +1,6 @@ +-- AlterEnum +ALTER TYPE "TaskRunStatus" ADD VALUE 'DELAYED'; + +-- AlterTable +ALTER TABLE "TaskRun" ADD COLUMN "delayUntil" TIMESTAMP(3), +ADD COLUMN "queuedAt" TIMESTAMP(3); diff --git a/packages/database/prisma/migrations/20240630204935_add_ttl_schema_changes/migration.sql b/packages/database/prisma/migrations/20240630204935_add_ttl_schema_changes/migration.sql new file mode 100644 index 000000000..25089f05f --- /dev/null +++ b/packages/database/prisma/migrations/20240630204935_add_ttl_schema_changes/migration.sql @@ -0,0 +1,6 @@ +-- AlterEnum +ALTER TYPE "TaskRunStatus" ADD VALUE 'EXPIRED'; + +-- AlterTable +ALTER TABLE "TaskRun" ADD COLUMN "expiredAt" TIMESTAMP(3), +ADD COLUMN "ttl" TEXT; diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index 1a5a6801a..07b0dc9a1 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -1660,6 +1660,11 @@ model TaskRun { concurrencyKey String? + delayUntil DateTime? + queuedAt DateTime? + ttl String? + expiredAt DateTime? + batchItems BatchTaskRunItem[] dependency TaskRunDependency? CheckpointRestoreEvent CheckpointRestoreEvent[] @@ -1684,6 +1689,8 @@ model TaskRun { } enum TaskRunStatus { + /// Task has been scheduled to run in the future + DELAYED /// Task is waiting to be executed by a worker PENDING @@ -1719,6 +1726,9 @@ enum TaskRunStatus { /// Task has crashed and won't be retried, most likely the worker ran out of resources, e.g. memory or storage CRASHED + + // Task reached the ttl without being executed + EXPIRED } model TaskRunDependency { diff --git a/packages/trigger-sdk/src/v3/runs.ts b/packages/trigger-sdk/src/v3/runs.ts index e3c76e17c..5ad4a175e 100644 --- a/packages/trigger-sdk/src/v3/runs.ts +++ b/packages/trigger-sdk/src/v3/runs.ts @@ -1,4 +1,8 @@ -import type { ListProjectRunsQueryParams, ListRunsQueryParams } from "@trigger.dev/core/v3"; +import type { + ListProjectRunsQueryParams, + ListRunsQueryParams, + RescheduleRunRequestBody, +} from "@trigger.dev/core/v3"; import { ApiPromise, CanceledRunResponse, @@ -21,6 +25,7 @@ export const runs = { cancel: cancelRun, retrieve: retrieveRun, list: listRuns, + reschedule: rescheduleRun, poll, }; @@ -84,6 +89,19 @@ function cancelRun(runId: string): ApiPromise { return apiClient.cancelRun(runId); } +function rescheduleRun( + runId: string, + body: RescheduleRunRequestBody +): ApiPromise { + const apiClient = apiClientManager.client; + + if (!apiClient) { + throw apiClientMissingError(); + } + + return apiClient.rescheduleRun(runId, body); +} + export type PollOptions = { pollIntervalMs?: number }; async function poll | string>( diff --git a/packages/trigger-sdk/src/v3/shared.ts b/packages/trigger-sdk/src/v3/shared.ts index 9b2909d09..f2fbade41 100644 --- a/packages/trigger-sdk/src/v3/shared.ts +++ b/packages/trigger-sdk/src/v3/shared.ts @@ -154,6 +154,7 @@ export type TaskOptions< | "large-1x" | "large-2x"; }; + /** This gets called when a task is triggered. It's where you put the code you want to execute. * * @param payload - The payload that is passed to your task when it's triggered. This must be JSON serializable. @@ -363,10 +364,36 @@ export type TaskIdentifier = TTask extends Task( test: taskContext.ctx?.run.isTest, payloadType: payloadPacket.dataType, idempotencyKey: options?.idempotencyKey, + delay: options?.delay, + ttl: options?.ttl, }, }); @@ -846,6 +883,8 @@ export async function batchTrigger( test: taskContext.ctx?.run.isTest, payloadType: payloadPacket.dataType, idempotencyKey: item.options?.idempotencyKey, + delay: item.options?.delay, + ttl: item.options?.ttl, }, }; }) diff --git a/references/v3-catalog/src/clientUsage.ts b/references/v3-catalog/src/clientUsage.ts index 5fca6551d..9e0ced5d0 100644 --- a/references/v3-catalog/src/clientUsage.ts +++ b/references/v3-catalog/src/clientUsage.ts @@ -8,59 +8,74 @@ type createJsonHeroDocIdentifier = TaskIdentifier; // type createJsonHeroDocHandle = TaskOutputHandle; // retrieves the handle of the task async function main() { - const anyHandle = await tasks.trigger("create-jsonhero-doc", { - title: "Hello World", - content: { - message: "Hello, World!", + const anyHandle = await tasks.trigger( + "create-jsonhero-doc", + { + title: "Hello World", + content: { + message: "Hello, World!", + }, }, - }); + { + delay: "1m", + ttl: "1m", + } + ); const anyRun = await runs.retrieve(anyHandle); - console.log(`Run ${anyHandle.id} completed with output:`, anyRun.output); + console.log(`Run ${anyHandle.id} status: ${anyRun.status}, ttl: ${anyRun.ttl}`); - const handle = await tasks.trigger("create-jsonhero-doc", { - title: "Hello World", - content: { - message: "Hello, World!", - }, - }); + await new Promise((resolve) => setTimeout(resolve, 121000)); // wait for 2 minutes - console.log(handle); + const expiredRun = await runs.retrieve(anyRun.id); - const completedRun = await runs.poll(handle, { pollIntervalMs: 100 }); + console.log( + `Run ${anyHandle.id} status: ${expiredRun.status}, expired at: ${expiredRun.expiredAt}` + ); - console.log(`Run ${handle.id} completed with output:`, completedRun.output); + // const handle = await tasks.trigger("create-jsonhero-doc", { + // title: "Hello World", + // content: { + // message: "Hello, World!", + // }, + // }); - const run = await tasks.triggerAndPoll("create-jsonhero-doc", { - title: "Hello World", - content: { - message: "Hello, World!", - }, - }); + // console.log(handle); - console.log(`Run ${run.id} completed with output: `, run.output); + // const completedRun = await runs.poll(handle, { pollIntervalMs: 100 }); - const batchHandle = await tasks.batchTrigger("create-jsonhero-doc", [ - { - payload: { - title: "Hello World", - content: { - message: "Hello, World!", - }, - }, - }, - { - payload: { - title: "Hello World 2", - content: { - message: "Hello, World 2!", - }, - }, - }, - ]); + // console.log(`Run ${handle.id} completed with output:`, completedRun.output); - const run2 = await runs.retrieve(batchHandle.runs[0]); + // const run = await tasks.triggerAndPoll("create-jsonhero-doc", { + // title: "Hello World", + // content: { + // message: "Hello, World!", + // }, + // }); + + // console.log(`Run ${run.id} completed with output: `, run.output); + + // const batchHandle = await tasks.batchTrigger("create-jsonhero-doc", [ + // { + // payload: { + // title: "Hello World", + // content: { + // message: "Hello, World!", + // }, + // }, + // }, + // { + // payload: { + // title: "Hello World 2", + // content: { + // message: "Hello, World 2!", + // }, + // }, + // }, + // ]); + + // const run2 = await runs.retrieve(batchHandle.runs[0]); } main().catch(console.error); diff --git a/references/v3-catalog/src/management.ts b/references/v3-catalog/src/management.ts index 964f7a6f4..05dceb8d1 100644 --- a/references/v3-catalog/src/management.ts +++ b/references/v3-catalog/src/management.ts @@ -239,5 +239,5 @@ async function doTriggerUnfriendlyTaskId() { // doListRuns().catch(console.error); // doScheduleLists().catch(console.error); // doSchedules().catch(console.error); -doEnvVars().catch(console.error); +// doEnvVars().catch(console.error); // doTriggerUnfriendlyTaskId().catch(console.error);