github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

WIP execution concurrency controls implemented via Redis

Browse Source 
- 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
This commit is contained in:
Eric Allam
2023-11-17 16:56:11 +00:00
committed by Eric Allam
parent e3c3aa91e1
commit 4af05a3908
24 changed files with 672 additions and 470 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/components/run/RunOverview.tsx
+10 -24
View File
@@ -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 (
<PageContainer>
<PageHeader>
@@ -115,15 +107,15 @@ export function RunOverview({ run, trigger, showRerun, paths }: RunOverviewProps
Test run
</span>
)}
{showRerun && hasFinished(run.status) && (
{showRerun && run.isFinished && (
<RerunPopover
runId={run.id}
runsPath={paths.runsPath}
environmentType={run.environment.type}
status={basicStatus}
status={run.basicStatus}
/>
)}
{!hasFinished(run.status) && <CancelRun runId={run.id} />}
{!run.isFinished && <CancelRun runId={run.id} />}
</PageButtons>
</PageTitleRow>
<PageInfoRow>
@@ -211,10 +203,10 @@ export function RunOverview({ run, trigger, showRerun, paths }: RunOverviewProps
);
})
) : (
<BlankTasks status={run.status} basicStatus={basicStatus} />
<BlankTasks status={run.basicStatus} />
)}
</div>
{(basicStatus === "COMPLETED" || basicStatus === "FAILED") && (
{(run.basicStatus === "COMPLETED" || run.basicStatus === "FAILED") && (
<div>
<Header2 className={cn("mb-2")}>Run Summary</Header2>
<RunPanel
@@ -285,14 +277,8 @@ export function RunOverview({ run, trigger, showRerun, paths }: RunOverviewProps
);
}
function BlankTasks({
status,
basicStatus,
}: {
status: JobRunStatus;
basicStatus: RunBasicStatus;
}) {
switch (basicStatus) {
function BlankTasks({ status }: { status: RunBasicStatus }) {
switch (status) {
default:
case "COMPLETED":
return <Paragraph variant="small">There were no tasks for this run.</Paragraph>;
apps/webapp/app/components/runs/RunStatuses.tsx
+19 -42
View File
@@ -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 (
<span className="flex items-center gap-1">
@@ -40,49 +28,25 @@ export function RunStatusIcon({ status, className }: { status: JobRunStatus; cla
case "SUCCESS":
return <CheckCircleIcon className={cn(runStatusClassNameColor(status), className)} />;
case "PENDING":
return <ClockIcon className={cn(runStatusClassNameColor(status), className)} />;
case "QUEUED":
return <ClockIcon className={cn(runStatusClassNameColor(status), className)} />;
case "PREPROCESSING":
case "STARTED":
case "WAITING_TO_CONTINUE":
case "WAITING_TO_EXECUTE":
case "EXECUTING":
return <Spinner className={cn(runStatusClassNameColor(status), className)} />;
case "FAILURE":
return <XCircleIcon className={cn(runStatusClassNameColor(status), className)} />;
case "TIMED_OUT":
return <ExclamationTriangleIcon className={cn(runStatusClassNameColor(status), className)} />;
case "UNRESOLVED_AUTH":
case "FAILURE":
case "ABORTED":
case "INVALID_PAYLOAD":
return <XCircleIcon className={cn(runStatusClassNameColor(status), className)} />;
case "WAITING_ON_CONNECTIONS":
return <WrenchIcon className={cn(runStatusClassNameColor(status), className)} />;
case "ABORTED":
return <XCircleIcon className={cn(runStatusClassNameColor(status), className)} />;
case "PREPROCESSING":
return <Spinner className={cn(runStatusClassNameColor(status), className)} />;
case "CANCELED":
return <NoSymbolIcon className={cn(runStatusClassNameColor(status), className)} />;
}
}
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}`);
}
}
}
apps/webapp/app/models/jobRun.server.ts
+45
View File
@@ -0,0 +1,45 @@
import type { JobRun, JobRunStatus } from "@trigger.dev/database";
const COMPLETED_STATUSES: Array<JobRun["status"]> = [
"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}`);
}
}
}
apps/webapp/app/models/jobRunExecution.server.ts
-47
View File
@@ -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,
});
}
apps/webapp/app/platform/zodWorker.server.ts
+184 -2
View File
@@ -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<any, any>;
@@ -103,6 +104,7 @@ export type ZodWorkerOptions<TMessageCatalog extends MessageCatalogSchema> = {
cleanup?: ZodWorkerCleanupOptions;
reporter?: ZodWorkerReporter;
shutdownTimeoutInMs?: number;
rateLimiter?: GraphileRateLimiter;
};
export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
@@ -115,6 +117,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
#runner?: GraphileRunner;
#cleanup: ZodWorkerCleanupOptions | undefined;
#reporter?: ZodWorkerReporter;
#rateLimiter?: GraphileRateLimiter;
#shutdownTimeoutInMs?: number;
#shuttingDown = false;
@@ -127,6 +130,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
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<TMessageCatalog extends MessageCatalogSchema> {
noHandleSignals: true,
taskList: this.#createTaskListFromTasks(),
parsedCronItems,
forbiddenFlags: this.#rateLimiter?.forbiddenFlags.bind(this.#rateLimiter),
});
if (!this.#runner) {
@@ -379,7 +384,11 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
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<T extends object>(obj: T): T {
}
return obj;
}
export interface GraphileRateLimiter {
forbiddenFlags(): Promise<string[]>;
wrapTask(t: Task): Task;
}
declare module "ioredis" {
interface RedisCommander<Context> {
beforeTask(
setKey: string,
maxSizeKey: string,
forbiddenFlagsKey: string,
jobId: string,
forbiddenFlag: string,
callback?: Callback<string>
): Result<string, Context>;
afterTask(
setKey: string,
maxSizeKey: string,
forbiddenFlagsKey: string,
jobId: string,
forbiddenFlag: string,
callback?: Callback<string>
): Result<string, Context>;
}
}
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<string[]> {
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}`;
}
}
apps/webapp/app/presenters/RunPresenter.server.ts
+3
View File
@@ -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,
apps/webapp/app/services/runs/cancelRun.server.ts
+4 -3
View File
@@ -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;
apps/webapp/app/services/runs/continueRun.server.ts
+2 -5
View File
@@ -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 }
);
apps/webapp/app/services/runs/createRun.server.ts
+1 -1
View File
@@ -101,7 +101,7 @@ export class CreateRunService {
{
id: run.id,
},
{ tx }
{ tx, queueName: `startRun:${run.jobId}` }
);
return run;
apps/webapp/app/services/runs/performRunExecutionV3.server.ts
+106 -302
View File
@@ -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<Awaited<ReturnType<typeof findRun>>>;
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<string, any>): Promise<void> {
async #failRunExecutionWithRetry(run: FoundRun, output: Record<string, any>): Promise<void> {
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<string, any>,
status: "FAILURE" | "ABORTED" | "TIMED_OUT" | "UNRESOLVED_AUTH" | "INVALID_PAYLOAD" = "FAILURE",
durationInMs: number = 0
): Promise<void> {
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 {
apps/webapp/app/services/runs/resumeRun.server.ts
+138
View File
@@ -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<Awaited<ReturnType<typeof findRun>>>;
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,
},
});
}
apps/webapp/app/services/runs/startRun.server.ts
+12 -32
View File
@@ -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<Awaited<ReturnType<typeof findRun>>>;
type RunConnectionsByKey = Awaited<ReturnType<typeof createRunConnections>>;
@@ -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) {
apps/webapp/app/services/tasks/resumeTask.server.ts
+3 -6
View File
@@ -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<ReturnType<typeof findTask>>;
@@ -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) {
apps/webapp/app/services/tasks/runTask.server.ts
+2 -2
View File
@@ -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"
apps/webapp/app/services/worker.server.ts
+15 -2
View File
@@ -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
apps/webapp/package.json
+1
View File
@@ -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",
docker/dev-compose.yml
+19
View File
@@ -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
docker/docker-compose.yml
+19
View File
@@ -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
packages/core/src/schemas/api.ts
-1
View File
@@ -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(),
});
packages/core/src/schemas/runs.ts
+3
View File
@@ -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({
packages/database/prisma/migrations/20231117145312_add_additional_run_statuses/migration.sql
+11
View File
@@ -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';
packages/database/prisma/schema.prisma
+3
View File
@@ -776,6 +776,9 @@ enum JobRunStatus {
WAITING_ON_CONNECTIONS
PREPROCESSING
STARTED
EXECUTING
WAITING_TO_CONTINUE
WAITING_TO_EXECUTE
SUCCESS
FAILURE
TIMED_OUT
perf/src/trigger.ts
+15 -1
View File
@@ -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" }
);
},
});
pnpm-lock.yaml
Generated
+57
View File
@@ -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'}