From 4af05a3908fb9efba7119dbb32d6860cddb41763 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 17 Nov 2023 16:56:11 +0000 Subject: [PATCH] WIP execution concurrency controls implemented via Redis - Split up resuming a run and executing a run - Added some new statuses to better show what is going on in a run - Removed preprocessing runs --- .../webapp/app/components/run/RunOverview.tsx | 34 +- .../app/components/runs/RunStatuses.tsx | 61 +-- apps/webapp/app/models/jobRun.server.ts | 45 ++ .../app/models/jobRunExecution.server.ts | 47 -- apps/webapp/app/platform/zodWorker.server.ts | 186 +++++++- .../app/presenters/RunPresenter.server.ts | 3 + .../app/services/runs/cancelRun.server.ts | 7 +- .../app/services/runs/continueRun.server.ts | 7 +- .../app/services/runs/createRun.server.ts | 2 +- .../runs/performRunExecutionV3.server.ts | 408 +++++------------- .../app/services/runs/resumeRun.server.ts | 138 ++++++ .../app/services/runs/startRun.server.ts | 44 +- .../app/services/tasks/resumeTask.server.ts | 9 +- .../app/services/tasks/runTask.server.ts | 4 +- apps/webapp/app/services/worker.server.ts | 17 +- apps/webapp/package.json | 1 + docker/dev-compose.yml | 19 + docker/docker-compose.yml | 19 + packages/core/src/schemas/api.ts | 1 - packages/core/src/schemas/runs.ts | 3 + .../migration.sql | 11 + packages/database/prisma/schema.prisma | 3 + perf/src/trigger.ts | 16 +- pnpm-lock.yaml | 57 +++ 24 files changed, 672 insertions(+), 470 deletions(-) create mode 100644 apps/webapp/app/models/jobRun.server.ts delete mode 100644 apps/webapp/app/models/jobRunExecution.server.ts create mode 100644 apps/webapp/app/services/runs/resumeRun.server.ts create mode 100644 packages/database/prisma/migrations/20231117145312_add_additional_run_statuses/migration.sql diff --git a/apps/webapp/app/components/run/RunOverview.tsx b/apps/webapp/app/components/run/RunOverview.tsx index 96f15f53a..2b32f064c 100644 --- a/apps/webapp/app/components/run/RunOverview.tsx +++ b/apps/webapp/app/components/run/RunOverview.tsx @@ -10,9 +10,10 @@ import { useNavigate, useNavigation, } from "@remix-run/react"; -import { JobRunStatus, RuntimeEnvironmentType } from "@trigger.dev/database"; +import { RuntimeEnvironmentType } from "@trigger.dev/database"; import { useMemo } from "react"; import { usePathName } from "~/hooks/usePathName"; +import type { RunBasicStatus } from "~/models/jobRun.server"; import { ViewRun } from "~/presenters/RunPresenter.server"; import { cancelSchema } from "~/routes/resources.runs.$runId.cancel"; import { schema } from "~/routes/resources.runs.$runId.rerun"; @@ -38,14 +39,7 @@ import { } from "../primitives/PageHeader"; import { Paragraph } from "../primitives/Paragraph"; import { Popover, PopoverContent, PopoverTrigger } from "../primitives/Popover"; -import { - RunBasicStatus, - RunStatusIcon, - RunStatusLabel, - hasFinished, - runBasicStatus, - runStatusTitle, -} from "../runs/RunStatuses"; +import { RunStatusIcon, RunStatusLabel, runStatusTitle } from "../runs/RunStatuses"; import { RunPanel, RunPanelBody, @@ -95,8 +89,6 @@ export function RunOverview({ run, trigger, showRerun, paths }: RunOverviewProps } }, [pathName]); - const basicStatus = runBasicStatus(run.status); - return ( @@ -115,15 +107,15 @@ export function RunOverview({ run, trigger, showRerun, paths }: RunOverviewProps Test run )} - {showRerun && hasFinished(run.status) && ( + {showRerun && run.isFinished && ( )} - {!hasFinished(run.status) && } + {!run.isFinished && } @@ -211,10 +203,10 @@ export function RunOverview({ run, trigger, showRerun, paths }: RunOverviewProps ); }) ) : ( - + )} - {(basicStatus === "COMPLETED" || basicStatus === "FAILED") && ( + {(run.basicStatus === "COMPLETED" || run.basicStatus === "FAILED") && (
Run Summary There were no tasks for this run.; diff --git a/apps/webapp/app/components/runs/RunStatuses.tsx b/apps/webapp/app/components/runs/RunStatuses.tsx index 828fe50b4..087006ba2 100644 --- a/apps/webapp/app/components/runs/RunStatuses.tsx +++ b/apps/webapp/app/components/runs/RunStatuses.tsx @@ -10,18 +10,6 @@ import type { JobRunStatus } from "@trigger.dev/database"; import { cn } from "~/utils/cn"; import { Spinner } from "../primitives/Spinner"; -export function hasFinished(status: JobRunStatus): boolean { - return ( - status === "SUCCESS" || - status === "FAILURE" || - status === "ABORTED" || - status === "TIMED_OUT" || - status === "CANCELED" || - status === "UNRESOLVED_AUTH" || - status === "INVALID_PAYLOAD" - ); -} - export function RunStatus({ status }: { status: JobRunStatus }) { return ( @@ -40,49 +28,25 @@ export function RunStatusIcon({ status, className }: { status: JobRunStatus; cla case "SUCCESS": return ; case "PENDING": - return ; case "QUEUED": return ; + case "PREPROCESSING": case "STARTED": + case "WAITING_TO_CONTINUE": + case "WAITING_TO_EXECUTE": + case "EXECUTING": return ; - case "FAILURE": - return ; case "TIMED_OUT": return ; case "UNRESOLVED_AUTH": + case "FAILURE": + case "ABORTED": case "INVALID_PAYLOAD": return ; case "WAITING_ON_CONNECTIONS": return ; - case "ABORTED": - return ; - case "PREPROCESSING": - return ; case "CANCELED": return ; - } -} - -export type RunBasicStatus = "WAITING" | "PENDING" | "RUNNING" | "COMPLETED" | "FAILED"; - -export function runBasicStatus(status: JobRunStatus): RunBasicStatus { - switch (status) { - case "WAITING_ON_CONNECTIONS": - case "QUEUED": - case "PREPROCESSING": - case "PENDING": - return "PENDING"; - case "STARTED": - return "RUNNING"; - case "FAILURE": - case "TIMED_OUT": - case "UNRESOLVED_AUTH": - case "CANCELED": - case "ABORTED": - case "INVALID_PAYLOAD": - return "FAILED"; - case "SUCCESS": - return "COMPLETED"; default: { const _exhaustiveCheck: never = status; throw new Error(`Non-exhaustive match for value: ${status}`); @@ -100,6 +64,12 @@ export function runStatusTitle(status: JobRunStatus): string { return "In progress"; case "QUEUED": return "Queued"; + case "EXECUTING": + return "Executing"; + case "WAITING_TO_CONTINUE": + return "Waiting"; + case "WAITING_TO_EXECUTE": + return "Queued"; case "FAILURE": return "Failed"; case "TIMED_OUT": @@ -130,6 +100,9 @@ export function runStatusClassNameColor(status: JobRunStatus): string { case "PENDING": return "text-slate-500"; case "STARTED": + case "EXECUTING": + case "WAITING_TO_CONTINUE": + case "WAITING_TO_EXECUTE": return "text-blue-500"; case "QUEUED": return "text-amber-300"; @@ -147,5 +120,9 @@ export function runStatusClassNameColor(status: JobRunStatus): string { return "text-blue-500"; case "CANCELED": return "text-slate-500"; + default: { + const _exhaustiveCheck: never = status; + throw new Error(`Non-exhaustive match for value: ${status}`); + } } } diff --git a/apps/webapp/app/models/jobRun.server.ts b/apps/webapp/app/models/jobRun.server.ts new file mode 100644 index 000000000..537879ce2 --- /dev/null +++ b/apps/webapp/app/models/jobRun.server.ts @@ -0,0 +1,45 @@ +import type { JobRun, JobRunStatus } from "@trigger.dev/database"; + +const COMPLETED_STATUSES: Array = [ + "CANCELED", + "ABORTED", + "SUCCESS", + "TIMED_OUT", + "INVALID_PAYLOAD", + "FAILURE", + "UNRESOLVED_AUTH", +]; + +export function isRunCompleted(status: JobRunStatus) { + return COMPLETED_STATUSES.includes(status); +} + +export type RunBasicStatus = "WAITING" | "PENDING" | "RUNNING" | "COMPLETED" | "FAILED"; + +export function runBasicStatus(status: JobRunStatus): RunBasicStatus { + switch (status) { + case "WAITING_ON_CONNECTIONS": + case "QUEUED": + case "PREPROCESSING": + case "PENDING": + return "PENDING"; + case "STARTED": + case "EXECUTING": + case "WAITING_TO_CONTINUE": + case "WAITING_TO_EXECUTE": + return "RUNNING"; + case "FAILURE": + case "TIMED_OUT": + case "UNRESOLVED_AUTH": + case "CANCELED": + case "ABORTED": + case "INVALID_PAYLOAD": + return "FAILED"; + case "SUCCESS": + return "COMPLETED"; + default: { + const _exhaustiveCheck: never = status; + throw new Error(`Non-exhaustive match for value: ${status}`); + } + } +} diff --git a/apps/webapp/app/models/jobRunExecution.server.ts b/apps/webapp/app/models/jobRunExecution.server.ts deleted file mode 100644 index c22cc5846..000000000 --- a/apps/webapp/app/models/jobRunExecution.server.ts +++ /dev/null @@ -1,47 +0,0 @@ -import { JobRun } from "@trigger.dev/database"; -import { PrismaClientOrTransaction } from "~/db.server"; -import { executionWorker } from "~/services/worker.server"; - -export async function dequeueRunExecutionV2(run: JobRun, tx: PrismaClientOrTransaction) { - return await executionWorker.dequeue(`job_run:${run.id}`, { - tx, - }); -} - -export type EnqueueRunExecutionV3Options = { - runAt?: Date; - skipRetrying?: boolean; -}; - -export async function enqueueRunExecutionV3( - run: JobRun, - tx: PrismaClientOrTransaction, - options: EnqueueRunExecutionV3Options = {} -) { - const reason = run.status === "PREPROCESSING" ? "PREPROCESS" : "EXECUTE_JOB"; - - return await executionWorker.enqueue( - "performRunExecutionV3", - { - id: run.id, - reason: reason, - }, - { - tx, - runAt: options.runAt, - queueName: `job_run:${run.id}`, - jobKey: `job_run:${reason}:${run.id}`, - maxAttempts: options.skipRetrying ? 1 : undefined, - } - ); -} - -export async function dequeueRunExecutionV3(run: JobRun, tx: PrismaClientOrTransaction) { - await executionWorker.dequeue(`job_run:EXECUTE_JOB:${run.id}`, { - tx, - }); - - await executionWorker.dequeue(`job_run:PREPROCESS:${run.id}`, { - tx, - }); -} diff --git a/apps/webapp/app/platform/zodWorker.server.ts b/apps/webapp/app/platform/zodWorker.server.ts index f917645d3..635626024 100644 --- a/apps/webapp/app/platform/zodWorker.server.ts +++ b/apps/webapp/app/platform/zodWorker.server.ts @@ -14,7 +14,8 @@ import { run as graphileRun, parseCronItems } from "graphile-worker"; import omit from "lodash.omit"; import { z } from "zod"; import { PrismaClient, PrismaClientOrTransaction } from "~/db.server"; -import { workerLogger as logger, trace } from "~/services/logger.server"; +import { workerLogger as logger, trace, workerLogger } from "~/services/logger.server"; +import { Callback, Redis, RedisOptions, Result } from "ioredis"; export interface MessageCatalogSchema { [key: string]: z.ZodFirstPartySchemaTypes | z.ZodDiscriminatedUnion; @@ -103,6 +104,7 @@ export type ZodWorkerOptions = { cleanup?: ZodWorkerCleanupOptions; reporter?: ZodWorkerReporter; shutdownTimeoutInMs?: number; + rateLimiter?: GraphileRateLimiter; }; export class ZodWorker { @@ -115,6 +117,7 @@ export class ZodWorker { #runner?: GraphileRunner; #cleanup: ZodWorkerCleanupOptions | undefined; #reporter?: ZodWorkerReporter; + #rateLimiter?: GraphileRateLimiter; #shutdownTimeoutInMs?: number; #shuttingDown = false; @@ -127,6 +130,7 @@ export class ZodWorker { this.#recurringTasks = options.recurringTasks; this.#cleanup = options.cleanup; this.#reporter = options.reporter; + this.#rateLimiter = options.rateLimiter; this.#shutdownTimeoutInMs = options.shutdownTimeoutInMs ?? 60000; // default to 60 seconds } @@ -150,6 +154,7 @@ export class ZodWorker { noHandleSignals: true, taskList: this.#createTaskListFromTasks(), parsedCronItems, + forbiddenFlags: this.#rateLimiter?.forbiddenFlags.bind(this.#rateLimiter), }); if (!this.#runner) { @@ -379,7 +384,11 @@ export class ZodWorker { return this.#handleMessage(key, payload, helpers); }; - taskList[key] = task; + if (this.#rateLimiter) { + taskList[key] = this.#rateLimiter.wrapTask(task); + } else { + taskList[key] = task; + } } for (const [key] of Object.entries(this.#recurringTasks ?? {})) { @@ -647,3 +656,176 @@ function removeUndefinedKeys(obj: T): T { } return obj; } + +export interface GraphileRateLimiter { + forbiddenFlags(): Promise; + wrapTask(t: Task): Task; +} + +declare module "ioredis" { + interface RedisCommander { + beforeTask( + setKey: string, + maxSizeKey: string, + forbiddenFlagsKey: string, + jobId: string, + forbiddenFlag: string, + callback?: Callback + ): Result; + + afterTask( + setKey: string, + maxSizeKey: string, + forbiddenFlagsKey: string, + jobId: string, + forbiddenFlag: string, + callback?: Callback + ): Result; + } +} + +export type RedisGraphileRateLimiterOptions = { + redis: RedisOptions; + prefix?: string; +}; + +// TODO: we need to somehow seed and update the rate limit for each flag in Redis +export class RedisGraphileRateLimiter implements GraphileRateLimiter { + private redis: Redis; + private prefix: string; + + constructor(options?: RedisGraphileRateLimiterOptions) { + this.redis = new Redis(options?.redis ?? {}); + this.prefix = options?.prefix ?? "tr:gw"; + + this.redis.defineCommand("beforeTask", { + numberOfKeys: 3, + lua: ` +local setKey = KEYS[1] +local maxSizeKey = KEYS[2] +local forbiddenFlagsKey = KEYS[3] +local jobId = ARGV[1] +local forbiddenFlag = ARGV[2] + +local maxSize = tonumber(redis.call('GET', maxSizeKey)) +if maxSize == nil then + return false -- maxSize not set +end + +redis.call('SADD', setKey, jobId) +local currentSize = redis.call('SCARD', setKey) + +if currentSize < maxSize then + redis.call('SREM', forbiddenFlagsKey, forbiddenFlag) + return true +else + redis.call('SADD', forbiddenFlagsKey, forbiddenFlag) + return false +end + `, + }); + + this.redis.defineCommand("afterTask", { + numberOfKeys: 3, + lua: ` +local setKey = KEYS[1] +local maxSizeKey = KEYS[2] +local forbiddenFlagsKey = KEYS[3] +local jobId = ARGV[1] +local forbiddenFlag = ARGV[2] + +local maxSize = tonumber(redis.call('GET', maxSizeKey)) +if maxSize == nil then + return false -- maxSize not set +end + +redis.call('SREM', setKey, jobId) +local currentSize = redis.call('SCARD', setKey) + +if currentSize < maxSize then + redis.call('SREM', forbiddenFlagsKey, forbiddenFlag) + return true +else + redis.call('SADD', forbiddenFlagsKey, forbiddenFlag) + return false +end + `, + }); + } + + async forbiddenFlags(): Promise { + return this.redis.smembers(this.#prefixKey("rl:forbiddenFlags")); + } + + // wrapTask + // Before the task is run we need to: + // get the max concurreny for the flag + // if there is no max concurreny for the flag, we can skip the rest of the steps + // for each flag with the prefix "rl:" + // add the job id to a redis set with the key "rl:flag" + // get the length of the set + // if the length of the set is greater or equal to the max concurrency + // we need to add the flag to the "forbidden flags" list + // After the task is run + // for each flag with the prefix "rl:" + // get the max concurreny for the flag + // if there is no max concurreny for the flag, we can skip the rest of the steps + // remove the job id from the redis set with the key "rl:flag" + // get the length of the set + // get the max concurreny for the flag + // if the length of the set is less than the max concurrency + // we need to remove the flag from the "forbidden flags" list + // we need to make sure that if there are any errors thrown in the task that we still perform the "after task" steps, and then rethrow the error + wrapTask(t: Task): Task { + return async (payload: unknown, helpers: JobHelpers) => { + const flags = Object.keys(helpers.job.flags ?? {}).filter((flag) => flag.startsWith("rl:")); + + if (flags.length === 0) { + return t(payload, helpers); + } + + // Before + // TODO: handle errors + const beforeResults = await Promise.all( + flags.map(async (flag) => { + const result = await this.redis.beforeTask( + this.#prefixKey(flag), + this.#prefixKey(`${flag}:maxSize`), + this.#prefixKey("rl:forbiddenFlags"), + String(helpers.job.id), + flag + ); + + return result; + }) + ); + + logger.debug("[rate-limiter] beforeTask results", { beforeResults, flags }); + + try { + await t(payload, helpers); + } finally { + // TODO: handle errors + const afterResults = await Promise.all( + flags.map(async (flag) => { + const result = await this.redis.afterTask( + this.#prefixKey(flag), + this.#prefixKey(`${flag}:maxSize`), + this.#prefixKey("rl:forbiddenFlags"), + String(helpers.job.id), + flag + ); + + return result; + }) + ); + + logger.debug("[rate-limiter] afterTask results", { afterResults, flags }); + } + }; + } + + #prefixKey(key: string): string { + return `${this.prefix}:${key}`; + } +} diff --git a/apps/webapp/app/presenters/RunPresenter.server.ts b/apps/webapp/app/presenters/RunPresenter.server.ts index 8b14e0e49..30f25f509 100644 --- a/apps/webapp/app/presenters/RunPresenter.server.ts +++ b/apps/webapp/app/presenters/RunPresenter.server.ts @@ -5,6 +5,7 @@ import { StyleSchema, } from "@trigger.dev/core"; import { PrismaClient, prisma } from "~/db.server"; +import { isRunCompleted, runBasicStatus } from "~/models/jobRun.server"; import { mergeProperties } from "~/utils/mergeProperties.server"; import { taskListToTree } from "~/utils/taskListToTree"; @@ -67,6 +68,8 @@ export class RunPresenter { id: run.id, number: run.number, status: run.status, + basicStatus: runBasicStatus(run.status), + isFinished: isRunCompleted(run.status), startedAt: run.startedAt, completedAt: run.completedAt, isTest: run.isTest, diff --git a/apps/webapp/app/services/runs/cancelRun.server.ts b/apps/webapp/app/services/runs/cancelRun.server.ts index f33457a81..4b4d5926a 100644 --- a/apps/webapp/app/services/runs/cancelRun.server.ts +++ b/apps/webapp/app/services/runs/cancelRun.server.ts @@ -1,6 +1,6 @@ import { PrismaClient, prisma } from "~/db.server"; -import { executionWorker } from "../worker.server"; -import { dequeueRunExecutionV3 } from "~/models/jobRunExecution.server"; +import { PerformRunExecutionV3Service } from "./performRunExecutionV3.server"; +import { ResumeRunService } from "./resumeRun.server"; export class CancelRunService { #prismaClient: PrismaClient; @@ -39,7 +39,8 @@ export class CancelRunService { }, }); - await dequeueRunExecutionV3(run, tx); + await PerformRunExecutionV3Service.dequeue(run, tx); + await ResumeRunService.dequeue(run, tx); }); } catch (error) { throw error; diff --git a/apps/webapp/app/services/runs/continueRun.server.ts b/apps/webapp/app/services/runs/continueRun.server.ts index c001b8a4f..982f0daac 100644 --- a/apps/webapp/app/services/runs/continueRun.server.ts +++ b/apps/webapp/app/services/runs/continueRun.server.ts @@ -1,6 +1,5 @@ -import { RuntimeEnvironmentType } from "@trigger.dev/database"; import { $transaction, Prisma, PrismaClient, prisma } from "~/db.server"; -import { enqueueRunExecutionV3 } from "~/models/jobRunExecution.server"; +import { ResumeRunService } from "./resumeRun.server"; const RESUMABLE_STATUSES = ["FAILURE", "TIMED_OUT", "UNRESOLVED_AUTH", "ABORTED", "CANCELED"]; @@ -39,9 +38,7 @@ export class ContinueRunService { }, }); - await enqueueRunExecutionV3(run, tx, { - skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT, - }); + await ResumeRunService.enqueue(run, tx); }, { timeout: 10000 } ); diff --git a/apps/webapp/app/services/runs/createRun.server.ts b/apps/webapp/app/services/runs/createRun.server.ts index 131835875..ee82ba887 100644 --- a/apps/webapp/app/services/runs/createRun.server.ts +++ b/apps/webapp/app/services/runs/createRun.server.ts @@ -101,7 +101,7 @@ export class CreateRunService { { id: run.id, }, - { tx } + { tx, queueName: `startRun:${run.jobId}` } ); return run; diff --git a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts index a9f836058..2cc4ee413 100644 --- a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts +++ b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts @@ -16,7 +16,7 @@ import { supportsFeature, } from "@trigger.dev/core"; import { BloomFilter } from "@trigger.dev/core-backend"; -import { RuntimeEnvironmentType, type Task } from "@trigger.dev/database"; +import { JobRun } from "@trigger.dev/database"; import { generateErrorMessage } from "zod-error"; import { eventRecordToApiJson } from "~/api.server"; import { @@ -26,7 +26,7 @@ import { } from "~/consts"; import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server"; import { detectResponseIsTimeout } from "~/models/endpoint.server"; -import { enqueueRunExecutionV3 } from "~/models/jobRunExecution.server"; +import { isRunCompleted } from "~/models/jobRun.server"; import { resolveRunConnections } from "~/models/runConnection.server"; import { prepareTasksForCaching, prepareTasksForCachingLegacy } from "~/models/task.server"; import { CompleteRunTaskService } from "~/routes/api.v1.runs.$runId.tasks.$id.complete"; @@ -36,8 +36,9 @@ import { EndpointApi } from "../endpointApi.server"; import { createExecutionEvent } from "../executions/createExecutionEvent.server"; import { logger } from "../logger.server"; import { ResumeTaskService } from "../tasks/resumeTask.server"; -import { workerQueue } from "../worker.server"; +import { executionWorker, workerQueue } from "../worker.server"; import { forceYieldCoordinator } from "./forceYieldCoordinator.server"; +import { ResumeRunService } from "./resumeRun.server"; type FoundRun = NonNullable>>; type FoundTask = FoundRun["tasks"][number]; @@ -74,206 +75,80 @@ export class PerformRunExecutionV3Service { return; } - switch (input.reason) { - case "PREPROCESS": { - await this.#executePreprocessing(run); - break; - } - case "EXECUTE_JOB": { - await this.#executeJob(run, input, driftInMs); - break; - } - } + await this.#executeJob(run, input, driftInMs); } - // Execute the preprocessing step of a run, which will send the payload to the endpoint and give the job - // an opportunity to generate run properties based on the payload. - // If the endpoint is not available, or the response is not ok, - // the run execution will be marked as failed and the run will start - async #executePreprocessing(run: FoundRun) { - const client = new EndpointApi(run.environment.apiKey, run.endpoint.url); - const event = eventRecordToApiJson(run.event); - - const { response, parser } = await client.preprocessRunRequest({ - event, - job: { - id: run.version.job.slug, - version: run.version.version, - }, - run: { + static async enqueue( + run: JobRun, + tx: PrismaClientOrTransaction, + options: { + runAt?: Date; + skipRetrying?: boolean; + } = {} + ) { + return await executionWorker.enqueue( + "performRunExecutionV3", + { id: run.id, - isTest: run.isTest, + reason: "EXECUTE_JOB", }, - environment: { - id: run.environment.id, - slug: run.environment.slug, - type: run.environment.type, - }, - organization: { - id: run.organization.id, - slug: run.organization.slug, - title: run.organization.title, - }, - account: run.externalAccount - ? { - id: run.externalAccount.identifier, - metadata: run.externalAccount.metadata, - } - : undefined, - }); - - if (!response) { - return await this.#failRunExecution(this.#prismaClient, "PREPROCESS", run, { - message: "Could not connect to the endpoint", - }); - } - - if (!response.ok) { - return await this.#failRunExecution(this.#prismaClient, "PREPROCESS", run, { - message: `Endpoint responded with ${response.status} status code`, - }); - } - - const rawBody = await response.text(); - const safeBody = safeJsonZodParse(parser, rawBody); - - if (!safeBody) { - return await this.#failRunExecution(this.#prismaClient, "PREPROCESS", run, { - message: "Endpoint responded with invalid JSON", - }); - } - - if (!safeBody.success) { - return await this.#failRunExecution(this.#prismaClient, "PREPROCESS", run, { - message: generateErrorMessage(safeBody.error.issues), - }); - } - - if (safeBody.data.abort) { - return this.#failRunExecution( - this.#prismaClient, - "PREPROCESS", - run, - { message: "Endpoint aborted the run" }, - "ABORTED" - ); - } else { - await $transaction(this.#prismaClient, async (tx) => { - await tx.jobRun.update({ - where: { - id: run.id, - }, - data: { - status: "STARTED", - startedAt: new Date(), - properties: safeBody.data.properties, - forceYieldImmediately: false, - }, - }); - - await enqueueRunExecutionV3(run, tx, { - skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT, - }); - }); - } + { + tx, + runAt: options.runAt, + queueName: `job_run:${run.id}`, + jobKey: `job_run:EXECUTE_JOB:${run.id}`, + maxAttempts: options.skipRetrying ? 1 : undefined, + flags: [`rl:executions:${run.organizationId}`], + priority: run.number, + } + ); } + + static async dequeue(run: JobRun, tx: PrismaClientOrTransaction) { + await executionWorker.dequeue(`job_run:EXECUTE_JOB:${run.id}`, { + tx, + }); + } + async #executeJob(run: FoundRun, input: PerformRunExecutionV3Input, driftInMs: number = 0) { try { - const { isRetry, resumeTaskId } = input; - - if (run.status === "CANCELED") { - await this.#cancelExecution(run); + if (isRunCompleted(run.status)) { return; } - try { - if ( - typeof process.env.BLOCKED_ORGS === "string" && - process.env.BLOCKED_ORGS.includes(run.organizationId) - ) { - logger.debug("Skipping execution for blocked org", { - orgId: run.organizationId, - }); - - await this.#prismaClient.jobRun.update({ - where: { - id: run.id, - }, - data: { - status: "CANCELED", - completedAt: new Date(), - }, - }); - - return; - } - } catch (e) {} - const client = new EndpointApi(run.environment.apiKey, run.endpoint.url); const event = eventRecordToApiJson(run.event); const startedAt = new Date(); - const { executionCount } = await this.#prismaClient.jobRun.update({ - where: { - id: run.id, - }, - data: { - status: run.status === "QUEUED" ? "STARTED" : run.status, - startedAt: run.startedAt ?? new Date(), - executionCount: { - increment: 1, - }, - }, - select: { - executionCount: true, - }, - }); - const connections = await resolveRunConnections(run.runConnections); if (!connections.success) { - return this.#failRunExecution(this.#prismaClient, "EXECUTE_JOB", run, { + return this.#failRunExecution(this.#prismaClient, run, { message: `Could not resolve all connections for run ${run.id}. This should not happen`, }); } - let resumedTask: Task | undefined; - - if (resumeTaskId) { - resumedTask = - (await this.#prismaClient.task.findUnique({ - where: { - id: resumeTaskId, - }, - })) ?? undefined; - - if (resumedTask) { - resumedTask = await this.#prismaClient.task.update({ - where: { - id: resumeTaskId, - }, - data: { - status: resumedTask.noop ? "COMPLETED" : "RUNNING", - completedAt: resumedTask.noop ? new Date() : undefined, - }, - }); - } - } - const sourceContext = RunSourceContextSchema.safeParse(run.event.sourceContext); const executionBody = await this.#createExecutionBody( run, - [run.tasks, resumedTask].flat().filter(Boolean), + run.tasks, startedAt, - isRetry, + false, connections.auth, event, sourceContext.success ? sourceContext.data : undefined ); - forceYieldCoordinator.registerRun(run.id); + await this.#prismaClient.jobRun.update({ + where: { + id: run.id, + }, + data: { + status: "EXECUTING", + }, + }); await createExecutionEvent({ eventType: "start", @@ -286,6 +161,9 @@ export class PerformRunExecutionV3Service { runId: run.id, }); + forceYieldCoordinator.registerRun(run.id); + + // TODO: add the ability to abort the execution from any server using Redis pub/sub const { response, parser, errorParser, headersParser, durationInMs } = await client.executeJobRequest(executionBody); @@ -303,7 +181,7 @@ export class PerformRunExecutionV3Service { forceYieldCoordinator.deregisterRun(run.id); if (!response) { - return await this.#failRunExecutionWithRetry({ + return await this.#failRunExecutionWithRetry(run, { message: `Connection could not be established to the endpoint (${run.endpoint.url})`, }); } @@ -393,14 +271,9 @@ export class PerformRunExecutionV3Service { if (errorBody && errorBody.success) { // Only retry if the error isn't a 4xx if (response.status >= 400 && response.status <= 499) { - return await this.#failRunExecution( - this.#prismaClient, - "EXECUTE_JOB", - run, - errorBody.data - ); + return await this.#failRunExecution(this.#prismaClient, run, errorBody.data); } else { - return await this.#failRunExecutionWithRetry(errorBody.data); + return await this.#failRunExecutionWithRetry(run, errorBody.data); } } @@ -408,7 +281,6 @@ export class PerformRunExecutionV3Service { if (response.status >= 400 && response.status <= 499 && response.status !== 408) { return await this.#failRunExecution( this.#prismaClient, - "EXECUTE_JOB", run, { message: `Endpoint responded with ${response.status} status code`, @@ -423,11 +295,10 @@ export class PerformRunExecutionV3Service { this.#prismaClient, run, input, - durationInMs, - executionCount + durationInMs ); } else { - return await this.#failRunExecutionWithRetry({ + return await this.#failRunExecutionWithRetry(run, { message: `Endpoint responded with ${response.status} status code`, }); } @@ -439,7 +310,6 @@ export class PerformRunExecutionV3Service { if (!safeBody) { return await this.#failRunExecution( this.#prismaClient, - "EXECUTE_JOB", run, { message: "Endpoint responded with invalid JSON", @@ -452,7 +322,6 @@ export class PerformRunExecutionV3Service { if (!safeBody.success) { return await this.#failRunExecution( this.#prismaClient, - "EXECUTE_JOB", run, { message: generateErrorMessage(safeBody.error.issues), @@ -491,7 +360,6 @@ export class PerformRunExecutionV3Service { break; } case "CANCELED": { - await this.#cancelExecution(run); break; } case "UNRESOLVED_AUTH_ERROR": { @@ -644,6 +512,9 @@ export class PerformRunExecutionV3Service { executionDuration: { increment: durationInMs, }, + executionCount: { + increment: 1, + }, }, }); @@ -661,17 +532,18 @@ export class PerformRunExecutionV3Service { run: FoundRun, data: RunJobResumeWithTask, durationInMs: number, - executionCount: number = 1 + executionCountIncrement: number = 1 ) { return await $transaction(this.#prismaClient, async (tx) => { await tx.jobRun.update({ where: { id: run.id }, data: { + status: "WAITING_TO_CONTINUE", executionDuration: { increment: durationInMs, }, executionCount: { - increment: executionCount, + increment: executionCountIncrement, }, }, }); @@ -744,7 +616,6 @@ export class PerformRunExecutionV3Service { case "ERROR": { return await this.#failRunExecution( this.#prismaClient, - "EXECUTE_JOB", run, childError.error ?? undefined, "FAILURE", @@ -754,7 +625,6 @@ export class PerformRunExecutionV3Service { case "INVALID_PAYLOAD": { return await this.#failRunExecution( this.#prismaClient, - "EXECUTE_JOB", run, childError.errors, "INVALID_PAYLOAD", @@ -774,7 +644,6 @@ export class PerformRunExecutionV3Service { case "UNRESOLVED_AUTH_ERROR": { return await this.#failRunExecution( this.#prismaClient, - "EXECUTE_JOB", run, childError.issues, "UNRESOLVED_AUTH", @@ -805,14 +674,7 @@ export class PerformRunExecutionV3Service { }); } - await this.#failRunExecution( - tx, - "EXECUTE_JOB", - execution, - data.error ?? undefined, - "FAILURE", - durationInMs - ); + await this.#failRunExecution(tx, execution, data.error ?? undefined, "FAILURE", durationInMs); }); } @@ -822,14 +684,7 @@ export class PerformRunExecutionV3Service { durationInMs: number ) { return await $transaction(this.#prismaClient, async (tx) => { - await this.#failRunExecution( - tx, - "EXECUTE_JOB", - execution, - data.issues, - "UNRESOLVED_AUTH", - durationInMs - ); + await this.#failRunExecution(tx, execution, data.issues, "UNRESOLVED_AUTH", durationInMs); }); } @@ -839,14 +694,7 @@ export class PerformRunExecutionV3Service { durationInMs: number ) { return await $transaction(this.#prismaClient, async (tx) => { - await this.#failRunExecution( - tx, - "EXECUTE_JOB", - execution, - data.errors, - "INVALID_PAYLOAD", - durationInMs - ); + await this.#failRunExecution(tx, execution, data.errors, "INVALID_PAYLOAD", durationInMs); }); } @@ -860,7 +708,6 @@ export class PerformRunExecutionV3Service { if (run.yieldedExecutions.length + 1 > MAX_RUN_YIELDED_EXECUTIONS) { return await this.#failRunExecution( tx, - "EXECUTE_JOB", run, { message: `Run has yielded too many times, the maximum is ${MAX_RUN_YIELDED_EXECUTIONS}`, @@ -875,6 +722,7 @@ export class PerformRunExecutionV3Service { id: run.id, }, data: { + status: "WAITING_TO_EXECUTE", executionDuration: { increment: durationInMs, }, @@ -892,9 +740,7 @@ export class PerformRunExecutionV3Service { }, }); - await enqueueRunExecutionV3(run, tx, { - skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT, - }); + await ResumeRunService.enqueue(run, tx); }); } @@ -910,6 +756,7 @@ export class PerformRunExecutionV3Service { id: run.id, }, data: { + status: "WAITING_TO_EXECUTE", executionDuration: { increment: durationInMs, }, @@ -933,9 +780,7 @@ export class PerformRunExecutionV3Service { }, }); - await enqueueRunExecutionV3(run, tx, { - skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT, - }); + await ResumeRunService.enqueue(run, tx); }); } @@ -981,9 +826,7 @@ export class PerformRunExecutionV3Service { output: data.output ? (JSON.parse(data.output) as any) : undefined, }); - await enqueueRunExecutionV3(run, tx, { - skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT, - }); + await ResumeRunService.enqueue(run, tx); }); } @@ -1035,6 +878,7 @@ export class PerformRunExecutionV3Service { status: "WAITING", run: { update: { + status: "WAITING_TO_CONTINUE", executionDuration: { increment: durationInMs, }, @@ -1054,8 +898,7 @@ export class PerformRunExecutionV3Service { prisma: PrismaClientOrTransaction, run: FoundRun, input: PerformRunExecutionV3Input, - durationInMs: number, - executionCount: number + durationInMs: number ) { await $transaction(prisma, async (tx) => { const executionDuration = run.executionDuration + durationInMs; @@ -1064,7 +907,6 @@ export class PerformRunExecutionV3Service { if (executionDuration >= run.organization.maximumExecutionTimePerRunInMs) { await this.#failRunExecution( tx, - "EXECUTE_JOB", run, { message: `Execution timed out after ${ @@ -1112,7 +954,6 @@ export class PerformRunExecutionV3Service { await this.#failRunExecution( tx, - "EXECUTE_JOB", run, { message: `Function timeout detected in ${ @@ -1147,102 +988,65 @@ export class PerformRunExecutionV3Service { }); // The run has timed out, so we need to enqueue a new execution - await enqueueRunExecutionV3(run, tx, { - skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT, - }); + await ResumeRunService.enqueue(run, tx); }); } - async #failRunExecutionWithRetry(output: Record): Promise { + async #failRunExecutionWithRetry(run: FoundRun, output: Record): Promise { + await this.#prismaClient.jobRun.update({ + where: { id: run.id }, + data: { + status: "WAITING_TO_EXECUTE", + }, + }); + throw new Error(JSON.stringify(output)); } async #failRunExecution( prisma: PrismaClientOrTransaction, - reason: "EXECUTE_JOB" | "PREPROCESS", run: FoundRun, output: Record, status: "FAILURE" | "ABORTED" | "TIMED_OUT" | "UNRESOLVED_AUTH" | "INVALID_PAYLOAD" = "FAILURE", durationInMs: number = 0 ): Promise { await $transaction(prisma, async (tx) => { - switch (reason) { - case "EXECUTE_JOB": { - // If the execution is an EXECUTE_JOB reason, we need to fail the run - await tx.jobRun.update({ - where: { id: run.id }, - data: { - completedAt: new Date(), - status, - output, - executionDuration: { - increment: durationInMs, - }, - tasks: { - updateMany: { - where: { - status: { - in: ["WAITING", "RUNNING", "PENDING"], - }, - }, - data: { - status: status === "TIMED_OUT" ? "CANCELED" : "ERRORED", - completedAt: new Date(), - }, + // If the execution is an EXECUTE_JOB reason, we need to fail the run + await tx.jobRun.update({ + where: { id: run.id }, + data: { + completedAt: new Date(), + status, + output, + executionDuration: { + increment: durationInMs, + }, + tasks: { + updateMany: { + where: { + status: { + in: ["WAITING", "RUNNING", "PENDING"], }, }, - forceYieldImmediately: false, - }, - }); - - await workerQueue.enqueue( - "deliverRunSubscriptions", - { - id: run.id, - }, - { tx } - ); - - break; - } - case "PREPROCESS": { - // If the status is ABORTED, we need to fail the run - if (status === "ABORTED") { - await tx.jobRun.update({ - where: { id: run.id }, data: { + status: status === "TIMED_OUT" ? "CANCELED" : "ERRORED", completedAt: new Date(), - status, - output, }, - }); - - break; - } - - await tx.jobRun.update({ - where: { - id: run.id, }, - data: { - status: "STARTED", - startedAt: new Date(), - }, - }); + }, + forceYieldImmediately: false, + }, + }); - await enqueueRunExecutionV3(run, tx, { - skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT, - }); - - break; - } - } + await workerQueue.enqueue( + "deliverRunSubscriptions", + { + id: run.id, + }, + { tx } + ); }); } - - async #cancelExecution(run: FoundRun) { - return; - } } function prepareNoOpTasksBloomFilter(possibleTasks: FoundTask[]): string { diff --git a/apps/webapp/app/services/runs/resumeRun.server.ts b/apps/webapp/app/services/runs/resumeRun.server.ts new file mode 100644 index 000000000..26e706d2c --- /dev/null +++ b/apps/webapp/app/services/runs/resumeRun.server.ts @@ -0,0 +1,138 @@ +import { JobRun, RuntimeEnvironmentType } from "@trigger.dev/database"; +import { PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server"; +import { workerQueue } from "../worker.server"; +import { PerformRunExecutionV3Service } from "./performRunExecutionV3.server"; + +type FoundRun = NonNullable>>; + +export class ResumeRunService { + #prismaClient: PrismaClient; + + constructor(prismaClient: PrismaClient = prisma) { + this.#prismaClient = prismaClient; + } + + public async call(id: string) { + const run = await findRun(this.#prismaClient, id); + + if (!run) { + return; + } + + switch (run.status) { + case "ABORTED": + case "CANCELED": + case "FAILURE": + case "INVALID_PAYLOAD": + case "SUCCESS": + case "TIMED_OUT": + case "UNRESOLVED_AUTH": { + return; + } + case "QUEUED": { + await this.#resumeQueuedRun(run); + break; + } + case "WAITING_TO_EXECUTE": { + await this.#executeRun(run); + break; + } + case "WAITING_TO_CONTINUE": + case "STARTED": { + await this.#resumeStartedRun(run); + break; + } + case "PENDING": + case "PREPROCESSING": { + await this.#resumePendingRun(run); + break; + } + case "EXECUTING": { + throw new Error("Cannot resume a run that is currently executing"); + } + case "WAITING_ON_CONNECTIONS": { + throw new Error("Cannot resume a run that is waiting on connections"); + } + default: { + const _exhaustiveCheck: never = run.status; + throw new Error(`Non-exhaustive match for value: ${run.status}`); + } + } + } + + async #resumeQueuedRun(run: FoundRun) { + await this.#prismaClient.jobRun.update({ + where: { + id: run.id, + }, + data: { + startedAt: run.startedAt ?? new Date(), + }, + }); + + await this.#executeRun(run); + } + + async #resumeStartedRun(run: FoundRun) { + await this.#prismaClient.jobRun.update({ + where: { + id: run.id, + }, + data: { + status: "WAITING_TO_EXECUTE", + }, + }); + + await this.#executeRun(run); + } + + async #resumePendingRun(run: FoundRun) { + await this.#prismaClient.jobRun.update({ + where: { + id: run.id, + }, + data: { + status: "QUEUED", + startedAt: new Date(), + }, + }); + + await this.#executeRun(run); + } + + async #executeRun(run: FoundRun) { + await PerformRunExecutionV3Service.enqueue(run, this.#prismaClient, { + skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT, + }); + } + + static async enqueue(run: JobRun, tx: PrismaClientOrTransaction, runAt?: Date) { + return await workerQueue.enqueue( + "resumeRun", + { + id: run.id, + }, + { + tx, + runAt: runAt, + queueName: `run_resume:${run.id}`, + jobKey: `run_resume:${run.id}`, + } + ); + } + + static async dequeue(run: JobRun, tx: PrismaClientOrTransaction) { + await workerQueue.dequeue(`run_resume:${run.id}`, { + tx, + }); + } +} + +async function findRun(prisma: PrismaClientOrTransaction, id: string) { + return await prisma.jobRun.findUnique({ + where: { id }, + include: { + environment: true, + }, + }); +} diff --git a/apps/webapp/app/services/runs/startRun.server.ts b/apps/webapp/app/services/runs/startRun.server.ts index 4ad39970b..cf5195497 100644 --- a/apps/webapp/app/services/runs/startRun.server.ts +++ b/apps/webapp/app/services/runs/startRun.server.ts @@ -1,13 +1,12 @@ import { - RuntimeEnvironmentType, type ConnectionType, type Integration, type IntegrationConnection, } from "@trigger.dev/database"; import type { PrismaClient, PrismaClientOrTransaction } from "~/db.server"; import { prisma } from "~/db.server"; -import { enqueueRunExecutionV3 } from "~/models/jobRunExecution.server"; import { workerQueue } from "../worker.server"; +import { ResumeRunService } from "./resumeRun.server"; type FoundRun = NonNullable>>; type RunConnectionsByKey = Awaited>; @@ -60,37 +59,18 @@ export class StartRunService { ) .filter(Boolean); - const updateRun = async () => { - if (run.preprocess) { - // Start the jobRun and increment the jobCount - return await this.#prismaClient.jobRun.update({ - where: { id }, - data: { - status: "PREPROCESSING", - runConnections: { - create: createRunConnections, - }, - }, - }); - } else { - return await this.#prismaClient.jobRun.update({ - where: { id }, - data: { - status: "QUEUED", - queuedAt: new Date(), - runConnections: { - create: createRunConnections, - }, - }, - }); - } - }; - - const updatedRun = await updateRun(); - - await enqueueRunExecutionV3(updatedRun, this.#prismaClient, { - skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT, + const updatedRun = await this.#prismaClient.jobRun.update({ + where: { id }, + data: { + status: "QUEUED", + queuedAt: new Date(), + runConnections: { + create: createRunConnections, + }, + }, }); + + await ResumeRunService.enqueue(updatedRun, this.#prismaClient); } async #handleMissingConnections(id: string, runConnectionsByKey: RunConnectionsByKey) { diff --git a/apps/webapp/app/services/tasks/resumeTask.server.ts b/apps/webapp/app/services/tasks/resumeTask.server.ts index c47b04791..03a59b473 100644 --- a/apps/webapp/app/services/tasks/resumeTask.server.ts +++ b/apps/webapp/app/services/tasks/resumeTask.server.ts @@ -1,8 +1,7 @@ import { PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server"; -import { workerQueue } from "../worker.server"; -import { enqueueRunExecutionV3 } from "~/models/jobRunExecution.server"; -import { RuntimeEnvironmentType } from "@trigger.dev/database"; import { logger } from "../logger.server"; +import { ResumeRunService } from "../runs/resumeRun.server"; +import { workerQueue } from "../worker.server"; type FoundTask = Awaited>; @@ -81,9 +80,7 @@ export class ResumeTaskService { } } - await enqueueRunExecutionV3(task.run, this.#prismaClient, { - skipRetrying: task.run.environment.type === RuntimeEnvironmentType.DEVELOPMENT, - }); + await ResumeRunService.enqueue(task.run, this.#prismaClient); } public static async enqueue(id: string, runAt?: Date, tx?: PrismaClientOrTransaction) { diff --git a/apps/webapp/app/services/tasks/runTask.server.ts b/apps/webapp/app/services/tasks/runTask.server.ts index eebce56cd..f0a8edd28 100644 --- a/apps/webapp/app/services/tasks/runTask.server.ts +++ b/apps/webapp/app/services/tasks/runTask.server.ts @@ -71,7 +71,7 @@ export class RunTaskService { status = "CANCELED"; } else { status = - delayUntilInFuture || callbackEnabled || taskBody.trigger + delayUntilInFuture || callbackEnabled ? "WAITING" : taskBody.noop ? "COMPLETED" @@ -180,7 +180,7 @@ export class RunTaskService { if (existingTask) { if (existingTask.status === "CANCELED") { const existingTaskStatus = - delayUntilInFuture || callbackEnabled || taskBody.trigger + delayUntilInFuture || callbackEnabled ? "WAITING" : taskBody.noop ? "COMPLETED" diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts index 79f299648..707d5f0d6 100644 --- a/apps/webapp/app/services/worker.server.ts +++ b/apps/webapp/app/services/worker.server.ts @@ -3,7 +3,7 @@ import { ScheduledPayloadSchema, addMissingVersionField } from "@trigger.dev/cor import { z } from "zod"; import { prisma } from "~/db.server"; import { env } from "~/env.server"; -import { ZodWorker } from "~/platform/zodWorker.server"; +import { RedisGraphileRateLimiter, ZodWorker } from "~/platform/zodWorker.server"; import { sendEmail } from "./email.server"; import { IndexEndpointService } from "./endpoints/indexEndpoint.server"; import { PerformEndpointIndexService } from "./endpoints/performEndpointIndexService"; @@ -26,6 +26,7 @@ import { DeliverRunSubscriptionsService } from "./runs/deliverRunSubscriptions.s import { ResumeTaskService } from "./tasks/resumeTask.server"; import { ExpireDispatcherService } from "./dispatchers/expireDispatcher.server"; import { InvokeEphemeralDispatcherService } from "./dispatchers/invokeEphemeralEventDispatcher.server"; +import { ResumeRunService } from "./runs/resumeRun.server"; const workerCatalog = { indexEndpoint: z.object({ @@ -95,6 +96,9 @@ const workerCatalog = { expireDispatcher: z.object({ id: z.string(), }), + resumeRun: z.object({ + id: z.string(), + }), }; const executionWorkerCatalog = { @@ -223,7 +227,6 @@ function getWorkerQueue() { "events.invokeDispatcher": { priority: 0, // smaller number = higher priority maxAttempts: 6, - queueName: (payload) => `dispatcher:${payload.id}`, // use a queue for a dispatcher so runs are created sequentially handler: async (payload, job) => { const service = new InvokeDispatcherService(); @@ -400,6 +403,15 @@ function getWorkerQueue() { handler: async (payload) => { const service = new ExpireDispatcherService(); + return await service.call(payload.id); + }, + }, + resumeRun: { + priority: 0, + maxAttempts: 10, + handler: async (payload, job) => { + const service = new ResumeRunService(); + return await service.call(payload.id); }, }, @@ -421,6 +433,7 @@ function getExecutionWorkerQueue() { }, shutdownTimeoutInMs: env.GRACEFUL_SHUTDOWN_TIMEOUT, schema: executionWorkerCatalog, + rateLimiter: new RedisGraphileRateLimiter(), tasks: { performRunExecutionV2: { priority: 0, // smaller number = higher priority diff --git a/apps/webapp/package.json b/apps/webapp/package.json index 50ad62218..295807e35 100644 --- a/apps/webapp/package.json +++ b/apps/webapp/package.json @@ -84,6 +84,7 @@ "highlight.run": "^7.3.4", "humanize-duration": "^3.27.3", "intl-parse-accept-language": "^1.0.0", + "ioredis": "^5.3.2", "isbot": "^3.6.5", "jsonpointer": "^5.0.1", "lodash.omit": "^4.5.0", diff --git a/docker/dev-compose.yml b/docker/dev-compose.yml index 015139b68..642510e28 100644 --- a/docker/dev-compose.yml +++ b/docker/dev-compose.yml @@ -2,6 +2,7 @@ version: "3" volumes: database-data: + redis-data: networks: app_network: @@ -42,3 +43,21 @@ services: PORT: 3030 networks: - app_network + + redis: + container_name: redis + image: redis:7 + restart: always + volumes: + - redis-data:/data + networks: + - app_network + ports: + - 6379:6379 + + redisinsight: + image: redislabs/redisinsight:latest + ports: + - "8001:8001" + volumes: + - redis-data:/redisinsight diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 41e935745..7c6f6cc58 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -3,6 +3,7 @@ version: "3" volumes: database-data: pgadmin-data: + redis-data: networks: app_network: @@ -41,3 +42,21 @@ services: - 5480:80 depends_on: - database + + redis: + container_name: redis + image: redis:7 + restart: always + volumes: + - redis-data:/data + networks: + - app_network + ports: + - 6379:6379 + + redisinsight: + image: redislabs/redisinsight:latest + ports: + - "8001:8001" + volumes: + - redis-data:/redisinsight diff --git a/packages/core/src/schemas/api.ts b/packages/core/src/schemas/api.ts index 582e2aceb..2159146e4 100644 --- a/packages/core/src/schemas/api.ts +++ b/packages/core/src/schemas/api.ts @@ -803,7 +803,6 @@ export const RunTaskOptionsSchema = z.object({ /** A No Operation means that the code won't be executed. This is used internally to implement features like [io.wait()](https://trigger.dev/docs/sdk/io/wait). */ noop: z.boolean().default(false), redact: RedactSchema.optional(), - trigger: TriggerMetadataSchema.optional(), parallel: z.boolean().optional(), }); diff --git a/packages/core/src/schemas/runs.ts b/packages/core/src/schemas/runs.ts index e62c1331b..8fc448b68 100644 --- a/packages/core/src/schemas/runs.ts +++ b/packages/core/src/schemas/runs.ts @@ -18,6 +18,9 @@ export const RunStatusSchema = z.union([ z.literal("CANCELED"), z.literal("UNRESOLVED_AUTH"), z.literal("INVALID_PAYLOAD"), + z.literal("EXECUTING"), + z.literal("WAITING_TO_CONTINUE"), + z.literal("WAITING_TO_EXECUTE"), ]); export const RunTaskSchema = z.object({ diff --git a/packages/database/prisma/migrations/20231117145312_add_additional_run_statuses/migration.sql b/packages/database/prisma/migrations/20231117145312_add_additional_run_statuses/migration.sql new file mode 100644 index 000000000..98fe1903c --- /dev/null +++ b/packages/database/prisma/migrations/20231117145312_add_additional_run_statuses/migration.sql @@ -0,0 +1,11 @@ +-- AlterEnum +-- This migration adds more than one value to an enum. +-- With PostgreSQL versions 11 and earlier, this is not possible +-- in a single migration. This can be worked around by creating +-- multiple migrations, each migration adding only one value to +-- the enum. + + +ALTER TYPE "JobRunStatus" ADD VALUE 'EXECUTING'; +ALTER TYPE "JobRunStatus" ADD VALUE 'WAITING_TO_CONTINUE'; +ALTER TYPE "JobRunStatus" ADD VALUE 'WAITING_TO_EXECUTE'; diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index 46b459d4c..2851393b6 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -776,6 +776,9 @@ enum JobRunStatus { WAITING_ON_CONNECTIONS PREPROCESSING STARTED + EXECUTING + WAITING_TO_CONTINUE + WAITING_TO_EXECUTE SUCCESS FAILURE TIMED_OUT diff --git a/perf/src/trigger.ts b/perf/src/trigger.ts index 6f8006fa5..de9a238d2 100644 --- a/perf/src/trigger.ts +++ b/perf/src/trigger.ts @@ -17,7 +17,7 @@ triggerClient.defineJob({ await io.runTask( "task-1", async (task) => { - await new Promise((resolve) => setTimeout(resolve, 2000)); + await new Promise((resolve) => setTimeout(resolve, 10000)); return { value: Math.random(), @@ -26,6 +26,8 @@ triggerClient.defineJob({ { name: "task 1" } ); + await io.wait("wait", 10); + await io.runTask( "task-2", async (task) => { @@ -35,5 +37,17 @@ triggerClient.defineJob({ }, { name: "task 2" } ); + + await io.runTask( + "task-3", + async (task) => { + await new Promise((resolve) => setTimeout(resolve, 20000)); + + return { + value: Math.random(), + }; + }, + { name: "task 3" } + ); }, }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6162633b8..7ad6314ee 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -175,6 +175,7 @@ importers: highlight.run: ^7.3.4 humanize-duration: ^3.27.3 intl-parse-accept-language: ^1.0.0 + ioredis: ^5.3.2 isbot: ^3.6.5 jsonpointer: ^5.0.1 lodash.omit: ^4.5.0 @@ -282,6 +283,7 @@ importers: highlight.run: 7.3.4 humanize-duration: 3.27.3 intl-parse-accept-language: 1.0.0 + ioredis: 5.3.2 isbot: 3.6.5 jsonpointer: 5.0.1 lodash.omit: 4.5.0 @@ -7896,6 +7898,10 @@ packages: /@humanwhocodes/object-schema/1.2.1: resolution: {integrity: sha512-ZnQMnLV4e7hDlUvw8H+U8ASL02SS2Gn6+9Ac3wGGLIe7+je2AeAOxPY+izIPJDfFDb7eDjev0Us8MO1iFRN8hA==} + /@ioredis/commands/1.2.0: + resolution: {integrity: sha512-Sx1pU8EM64o2BrqNpEO1CNLtKQwyhuXuqyfH7oGKCk+1a33d2r5saW8zNwm3j6BTExtjrv2BxTgzzkMwts6vGg==} + dev: false + /@isaacs/cliui/8.0.2: resolution: {integrity: sha512-O8jcjabXaleOG9DQ0+ARXWZBTfnP4WNAqzuiJK7ll44AmxGKv/J2M4TPjxjY3znBCfvBXFzucm1twdyFybFqEA==} engines: {node: '>=12'} @@ -17968,6 +17974,11 @@ packages: resolution: {integrity: sha512-rQ1+kcj+ttHG0MKVGBUXwayCCF1oh39BF5COIpRzuCEv8Mwjv0XucrI2ExNTOn9IlLifGClWQcU9BrZORvtw6Q==} engines: {node: '>=6'} + /cluster-key-slot/1.1.2: + resolution: {integrity: sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==} + engines: {node: '>=0.10.0'} + dev: false + /co/4.6.0: resolution: {integrity: sha512-QVb0dM5HvG+uaxitm8wONl7jltx8dqhfU33DcqtOZcLSVIKSDDLDi7+0LbAKiyI8hD9u42m2YxXSkMGWThaecQ==} engines: {iojs: '>= 1.0.0', node: '>= 0.12.0'} @@ -18900,6 +18911,11 @@ packages: /delegates/1.0.0: resolution: {integrity: sha512-bd2L678uiWATM6m5Z1VzNCErI3jiGzt6HGY8OVICs40JQq/HALfbyNJmp0UDakEY4pMMaN0Ly5om/B1VI/+xfQ==} + /denque/2.1.0: + resolution: {integrity: sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==} + engines: {node: '>=0.10'} + dev: false + /depd/2.0.0: resolution: {integrity: sha512-g7nH6P6dyDioJogAAGprGpCtVImJhpPk/roCzdb3fIh61/s/nPsfR6onyMwkCAR/OlC3yBC0lESvUoQEAssIrw==} engines: {node: '>= 0.8'} @@ -23212,6 +23228,23 @@ packages: loose-envify: 1.4.0 dev: false + /ioredis/5.3.2: + resolution: {integrity: sha512-1DKMMzlIHM02eBBVOFQ1+AolGjs6+xEcM4PDL7NqOS6szq7H9jSaEkIUH6/a5Hl241LzW6JLSiAbNvTQjUupUA==} + engines: {node: '>=12.22.0'} + dependencies: + '@ioredis/commands': 1.2.0 + cluster-key-slot: 1.1.2 + debug: 4.3.4 + denque: 2.1.0 + lodash.defaults: 4.2.0 + lodash.isarguments: 3.1.0 + redis-errors: 1.2.0 + redis-parser: 3.0.0 + standard-as-callback: 2.1.0 + transitivePeerDependencies: + - supports-color + dev: false + /ip/1.1.8: resolution: {integrity: sha512-PuExPYUiu6qMBQb4l06ecm6T6ujzhmh+MeJcW9wa89PoAz5pvd4zPgN5WJV104mb6S2T1AwNIAaB70JNrLQWhg==} @@ -24978,6 +25011,14 @@ packages: resolution: {integrity: sha512-FT1yDzDYEoYWhnSGnpE/4Kj1fLZkDFyqRb7fNt6FdYOSxlUWAtp42Eh6Wb0rGIv/m9Bgo7x4GhQbm5Ys4SG5ow==} dev: true + /lodash.defaults/4.2.0: + resolution: {integrity: sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==} + dev: false + + /lodash.isarguments/3.1.0: + resolution: {integrity: sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==} + dev: false + /lodash.isplainobject/4.0.6: resolution: {integrity: sha512-oSXzaWypCMHkPC3NvBEaPHf0KsA5mvPrOPgQWDsbg8n7orZ290M0BmC/jgRZ4vcJ6DTAhjrsSYgdsW/F+MFOBA==} dev: true @@ -29187,6 +29228,18 @@ packages: strip-indent: 3.0.0 dev: false + /redis-errors/1.2.0: + resolution: {integrity: sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==} + engines: {node: '>=4'} + dev: false + + /redis-parser/3.0.0: + resolution: {integrity: sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==} + engines: {node: '>=4'} + dependencies: + redis-errors: 1.2.0 + dev: false + /reduce-css-calc/2.1.8: resolution: {integrity: sha512-8liAVezDmUcH+tdzoEGrhfbGcP7nOV4NkGE3a74+qqvE7nt9i4sKLGBuZNOnpI4WiGksiNPklZxva80061QiPg==} dependencies: @@ -30712,6 +30765,10 @@ packages: get-source: 2.0.12 dev: true + /standard-as-callback/2.1.0: + resolution: {integrity: sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==} + dev: false + /static-extend/0.1.2: resolution: {integrity: sha512-72E9+uLc27Mt718pMHt9VMNiAL4LMsmDbBva8mxWUCkT07fSzEGMYUCk0XWY6lp0j6RBAG4cJ3mWuZv2OE3s0g==} engines: {node: '>=0.10.0'}