);
}
diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.statuses.$id.ts b/apps/webapp/app/routes/api.v1.runs.$runId.statuses.$id.ts
index e54753423..ca8bf9fb2 100644
--- a/apps/webapp/app/routes/api.v1.runs.$runId.statuses.$id.ts
+++ b/apps/webapp/app/routes/api.v1.runs.$runId.statuses.$id.ts
@@ -1,10 +1,7 @@
import type { ActionArgs } from "@remix-run/server-runtime";
import { json } from "@remix-run/server-runtime";
-import { TaskStatus } from "@trigger.dev/database";
import {
- RunTaskBodyOutput,
- RunTaskBodyOutputSchema,
- ServerTask,
+ JobRunStatusRecordSchema,
StatusHistory,
StatusHistorySchema,
StatusUpdate,
@@ -14,12 +11,8 @@ import {
} from "@trigger.dev/core";
import { z } from "zod";
import { $transaction, PrismaClient, prisma } from "~/db.server";
-import { taskWithAttemptsToServerTask } from "~/models/task.server";
import { authenticateApiRequest } from "~/services/apiAuth.server";
import { logger } from "~/services/logger.server";
-import { ulid } from "~/services/ulid.server";
-import { workerQueue } from "~/services/worker.server";
-import { JobRunStatusRecordSchema } from "@trigger.dev/core";
const ParamsSchema = z.object({
runId: z.string(),
diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts
index f116261cb..c84a7af3d 100644
--- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts
+++ b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts
@@ -3,7 +3,7 @@ import { json } from "@remix-run/server-runtime";
import type { CompleteTaskBodyOutput, ServerTask } from "@trigger.dev/core";
import { CompleteTaskBodyInputSchema } from "@trigger.dev/core";
import { z } from "zod";
-import { PrismaClient, prisma } from "~/db.server";
+import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
import { taskWithAttemptsToServerTask } from "~/models/task.server";
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { authenticateApiRequest } from "~/services/apiAuth.server";
@@ -72,9 +72,9 @@ export async function action({ request, params }: ActionArgs) {
}
export class CompleteRunTaskService {
- #prismaClient: PrismaClient;
+ #prismaClient: PrismaClientOrTransaction;
- constructor(prismaClient: PrismaClient = prisma) {
+ constructor(prismaClient: PrismaClientOrTransaction = prisma) {
this.#prismaClient = prismaClient;
}
@@ -86,7 +86,7 @@ export class CompleteRunTaskService {
): Promise {
// Using a transaction, we'll first check to see if the task already exists and return if if it does
// If it doesn't exist, we'll create it and return it
- const task = await this.#prismaClient.$transaction(async (tx) => {
+ const task = await $transaction(this.#prismaClient, async (tx) => {
const existingTask = await tx.task.findUnique({
where: {
id,
@@ -152,6 +152,7 @@ export class CompleteRunTaskService {
},
include: {
attempts: true,
+ run: true,
},
});
});
diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts
index 8bdfb4f44..56bef530e 100644
--- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts
+++ b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts
@@ -152,6 +152,7 @@ export class FailRunTaskService {
},
include: {
attempts: true,
+ run: true,
},
});
});
diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts
index 871a48d92..e19796b18 100644
--- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts
+++ b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts
@@ -184,6 +184,7 @@ export class RunTaskService {
},
include: {
attempts: true,
+ run: true,
},
});
diff --git a/apps/webapp/app/routes/resources.environments.$environmentParam.endpoint.$endpointParam.ts b/apps/webapp/app/routes/resources.environments.$environmentParam.endpoint.$endpointParam.ts
index df8fcf927..b4033d9dd 100644
--- a/apps/webapp/app/routes/resources.environments.$environmentParam.endpoint.$endpointParam.ts
+++ b/apps/webapp/app/routes/resources.environments.$environmentParam.endpoint.$endpointParam.ts
@@ -1,20 +1,26 @@
import { ActionArgs, json } from "@remix-run/server-runtime";
import { z } from "zod";
import { IndexEndpointService } from "~/services/endpoints/indexEndpoint.server";
-import { requireUserId } from "~/services/session.server";
+import { workerQueue } from "~/services/worker.server";
const ParamsSchema = z.object({
environmentParam: z.string(),
endpointParam: z.string(),
});
-export async function action({ request, params }: ActionArgs) {
- const userId = await requireUserId(request);
- const { environmentParam, endpointParam } = ParamsSchema.parse(params);
+export async function action({ params }: ActionArgs) {
+ const { endpointParam } = ParamsSchema.parse(params);
try {
const service = new IndexEndpointService();
- const result = await service.call(endpointParam, "MANUAL");
+ await service.call(endpointParam, "MANUAL");
+
+ // Enqueue the endpoint to be probed in 10 seconds
+ await workerQueue.enqueue(
+ "probeEndpoint",
+ { id: endpointParam },
+ { jobKey: `probe:${endpointParam}`, runAt: new Date(Date.now() + 10000) }
+ );
return json({ success: true });
} catch (e) {
diff --git a/apps/webapp/app/services/endpointApi.server.ts b/apps/webapp/app/services/endpointApi.server.ts
index 708a62ee7..af2c04d02 100644
--- a/apps/webapp/app/services/endpointApi.server.ts
+++ b/apps/webapp/app/services/endpointApi.server.ts
@@ -97,6 +97,7 @@ export class EndpointApi {
return {
...pongResponse.data,
triggerVersion: headers.data["trigger-version"],
+ triggerSdkVersion: headers.data["trigger-sdk-version"],
};
}
@@ -308,6 +309,27 @@ export class EndpointApi {
return validateResponse.data;
}
+
+ async probe(timeout: number) {
+ const startTimeInMs = performance.now();
+
+ const response = await safeFetch(this.url, {
+ method: "POST",
+ headers: {
+ "content-type": "application/json",
+ "x-trigger-api-key": this.apiKey,
+ "x-trigger-action": "PROBE_EXECUTION_TIMEOUT",
+ },
+ body: JSON.stringify({
+ timeout,
+ }),
+ });
+
+ return {
+ response,
+ durationInMs: Math.floor(performance.now() - startTimeInMs),
+ };
+ }
}
async function safeFetch(url: string, options: RequestInit) {
diff --git a/apps/webapp/app/services/endpoints/performEndpointIndexService.ts b/apps/webapp/app/services/endpoints/performEndpointIndexService.ts
index ca1c0bcd1..3457fa466 100644
--- a/apps/webapp/app/services/endpoints/performEndpointIndexService.ts
+++ b/apps/webapp/app/services/endpoints/performEndpointIndexService.ts
@@ -127,16 +127,21 @@ export class PerformEndpointIndexService {
}
const { jobs, sources, dynamicTriggers, dynamicSchedules } = bodyResult.data;
- const { "trigger-version": triggerVersion } = headerResult.data;
+ const { "trigger-version": triggerVersion, "trigger-sdk-version": triggerSdkVersion } =
+ headerResult.data;
const { endpoint } = endpointIndex;
- if (triggerVersion && triggerVersion !== endpoint.version) {
+ if (
+ (triggerVersion && triggerVersion !== endpoint.version) ||
+ (triggerSdkVersion && triggerSdkVersion !== endpoint.sdkVersion)
+ ) {
await this.#prismaClient.endpoint.update({
where: {
id: endpoint.id,
},
data: {
version: triggerVersion,
+ sdkVersion: triggerSdkVersion,
},
});
}
diff --git a/apps/webapp/app/services/endpoints/probeEndpoint.server.ts b/apps/webapp/app/services/endpoints/probeEndpoint.server.ts
new file mode 100644
index 000000000..ad8726d74
--- /dev/null
+++ b/apps/webapp/app/services/endpoints/probeEndpoint.server.ts
@@ -0,0 +1,64 @@
+import { MAX_RUN_CHUNK_EXECUTION_LIMIT, RESPONSE_TIMEOUT_STATUS_CODES } from "~/consts";
+import { prisma, PrismaClient } from "~/db.server";
+import { EndpointApi } from "../endpointApi.server";
+import { logger } from "../logger.server";
+import { detectResponseIsTimeout } from "~/models/endpoint.server";
+
+export class ProbeEndpointService {
+ #prismaClient: PrismaClient;
+
+ constructor(prismaClient: PrismaClient = prisma) {
+ this.#prismaClient = prismaClient;
+ }
+
+ public async call(id: string) {
+ const endpoint = await this.#prismaClient.endpoint.findUnique({
+ where: {
+ id,
+ },
+ include: {
+ environment: true,
+ },
+ });
+
+ if (!endpoint) {
+ return;
+ }
+
+ logger.debug(`Probing endpoint`, {
+ id,
+ });
+
+ const client = new EndpointApi(endpoint.environment.apiKey, endpoint.url);
+
+ const { response, durationInMs } = await client.probe(MAX_RUN_CHUNK_EXECUTION_LIMIT);
+
+ if (!response) {
+ return;
+ }
+
+ logger.debug(`Probing endpoint complete`, {
+ id,
+ durationInMs,
+ response: {
+ status: response.status,
+ headers: Object.fromEntries(response.headers.entries()),
+ },
+ });
+
+ // If the response is a 200, or it was a timeout, we can assume the endpoint is up and update the runChunkExecutionLimit
+ if (response.status === 200 || detectResponseIsTimeout(response)) {
+ await this.#prismaClient.endpoint.update({
+ where: {
+ id,
+ },
+ data: {
+ runChunkExecutionLimit: Math.min(
+ Math.max(durationInMs, 10000),
+ MAX_RUN_CHUNK_EXECUTION_LIMIT
+ ),
+ },
+ });
+ }
+ }
+}
diff --git a/apps/webapp/app/services/runs/forceYieldCoordinator.server.ts b/apps/webapp/app/services/runs/forceYieldCoordinator.server.ts
new file mode 100644
index 000000000..c6565424d
--- /dev/null
+++ b/apps/webapp/app/services/runs/forceYieldCoordinator.server.ts
@@ -0,0 +1,47 @@
+import { PrismaClient } from "@trigger.dev/database";
+import { prisma } from "~/db.server";
+import { logger } from "../logger.server";
+
+class ForceYieldCoordinator {
+ private inFlightRuns: Set = new Set();
+ private prismaClient: PrismaClient;
+
+ constructor(prismaClient: PrismaClient) {
+ this.prismaClient = prismaClient;
+
+ process.on("SIGTERM", this.handleForceYield);
+ }
+
+ // Add a run to the in-flight set
+ public registerRun(runId: string): void {
+ this.inFlightRuns.add(runId);
+ }
+
+ // Remove a run from the in-flight set
+ public deregisterRun(runId: string): void {
+ this.inFlightRuns.delete(runId);
+ }
+
+ // Handle forced yield on SIGTERM
+ private handleForceYield = async (): Promise => {
+ const runIds = Array.from(this.inFlightRuns);
+
+ const results = await this.prismaClient.jobRun.updateMany({
+ where: {
+ id: {
+ in: runIds,
+ },
+ forceYieldImmediately: false,
+ },
+ data: {
+ forceYieldImmediately: true,
+ },
+ });
+
+ logger.debug(
+ `ForceYieldCoordinator: ${results.count}/${runIds.length} runs set to immediately force yield`
+ );
+ };
+}
+
+export const forceYieldCoordinator = new ForceYieldCoordinator(prisma);
diff --git a/apps/webapp/app/services/runs/performRunExecutionV1.server.ts b/apps/webapp/app/services/runs/performRunExecutionV1.server.ts
deleted file mode 100644
index c67a9dd18..000000000
--- a/apps/webapp/app/services/runs/performRunExecutionV1.server.ts
+++ /dev/null
@@ -1,771 +0,0 @@
-import {
- CachedTaskSchema,
- RunJobError,
- RunJobInvalidPayloadError,
- RunJobResumeWithTask,
- RunJobRetryWithTask,
- RunJobSuccess,
- RunJobUnresolvedAuthError,
- RunSourceContextSchema,
-} from "@trigger.dev/core";
-import type { Task } from "@trigger.dev/database";
-import { generateErrorMessage } from "zod-error";
-import { eventRecordToApiJson } from "~/api.server";
-import { EXECUTE_JOB_RETRY_LIMIT } from "~/consts";
-import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
-import { enqueueRunExecutionV1 } from "~/models/jobRunExecution.server";
-import { resolveRunConnections } from "~/models/runConnection.server";
-import { formatError } from "~/utils/formatErrors.server";
-import { safeJsonZodParse } from "~/utils/json";
-import { EndpointApi } from "../endpointApi.server";
-import { logger } from "../logger.server";
-
-type FoundRunExecution = NonNullable>>;
-
-export class PerformRunExecutionV1Service {
- #prismaClient: PrismaClient;
-
- constructor(prismaClient: PrismaClient = prisma) {
- this.#prismaClient = prismaClient;
- }
-
- public async call(id: string) {
- const runExecution = await findRunExecution(this.#prismaClient, id);
-
- if (!runExecution) {
- return;
- }
-
- switch (runExecution.reason) {
- case "PREPROCESS": {
- await this.#executePreprocessing(runExecution);
- break;
- }
- case "EXECUTE_JOB": {
- await this.#executeJob(runExecution);
- break;
- }
- }
- }
-
- // 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(execution: FoundRunExecution) {
- const { run } = execution;
-
- const client = new EndpointApi(run.environment.apiKey, run.endpoint.url);
- const event = eventRecordToApiJson(run.event);
- const startedAt = new Date();
-
- await this.#prismaClient.jobRunExecution.update({
- where: {
- id: execution.id,
- },
- data: {
- status: "STARTED",
- startedAt,
- },
- });
-
- const { response, parser } = await client.preprocessRunRequest({
- event,
- job: {
- id: run.version.job.slug,
- version: run.version.version,
- },
- run: {
- id: run.id,
- isTest: run.isTest,
- },
- 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.#failRunExecutionWithRetry(execution, {
- message: "Could not connect to the endpoint",
- });
- }
-
- if (!response.ok) {
- return await this.#failRunExecutionWithRetry(execution, {
- 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, execution, {
- message: "Endpoint responded with invalid JSON",
- });
- }
-
- if (!safeBody.success) {
- return await this.#failRunExecution(this.#prismaClient, execution, {
- message: generateErrorMessage(safeBody.error.issues),
- });
- }
-
- if (safeBody.data.abort) {
- return this.#failRunExecution(
- this.#prismaClient,
- execution,
- { 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,
- },
- });
-
- await tx.jobRunExecution.update({
- where: {
- id: execution.id,
- },
- data: {
- status: "SUCCESS",
- completedAt: new Date(),
- },
- });
-
- const runExecution = await tx.jobRunExecution.create({
- data: {
- runId: run.id,
- reason: "EXECUTE_JOB",
- status: "PENDING",
- retryLimit: EXECUTE_JOB_RETRY_LIMIT,
- },
- });
-
- await enqueueRunExecutionV1(runExecution, run.queue.id, run.queue.maxJobs, tx);
- });
- }
- }
- async #executeJob(execution: FoundRunExecution) {
- const { run, isRetry } = execution;
-
- if (run.status === "CANCELED") {
- await this.#cancelExecution(execution);
- return;
- }
-
- const client = new EndpointApi(run.environment.apiKey, run.endpoint.url);
- const event = eventRecordToApiJson(run.event);
-
- const startedAt = new Date();
-
- await this.#prismaClient.jobRunExecution.update({
- where: {
- id: execution.id,
- },
- data: {
- status: "STARTED",
- startedAt,
- run: {
- update: {
- status: run.status === "QUEUED" ? "STARTED" : run.status,
- startedAt: run.startedAt ?? new Date(),
- },
- },
- },
- });
-
- const connections = await resolveRunConnections(run.runConnections);
-
- if (!connections.success) {
- return this.#failRunExecutionWithRetry(execution, {
- message: `Could not resolve all connections for run ${run.id}, attempting to retry`,
- });
- }
-
- let resumedTask: Task | undefined;
-
- if (execution.resumeTaskId) {
- resumedTask =
- (await this.#prismaClient.task.findUnique({
- where: {
- id: execution.resumeTaskId,
- },
- })) ?? undefined;
-
- if (resumedTask) {
- resumedTask = await this.#prismaClient.task.update({
- where: {
- id: execution.resumeTaskId,
- },
- data: {
- status: resumedTask.noop ? "COMPLETED" : "RUNNING",
- completedAt: resumedTask.noop ? new Date() : undefined,
- },
- });
- }
- }
-
- const sourceContext = RunSourceContextSchema.safeParse(run.event.sourceContext);
-
- const { response, parser, errorParser } = await client.executeJobRequest({
- event,
- job: {
- id: run.version.job.slug,
- version: run.version.version,
- },
- run: {
- id: run.id,
- isTest: run.isTest,
- startedAt,
- isRetry,
- },
- 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,
- connections: connections.auth,
- source: sourceContext.success ? sourceContext.data : undefined,
- tasks: [run.tasks, resumedTask]
- .flat()
- .filter(Boolean)
- .map((t) => CachedTaskSchema.parse(t)),
- yieldedExecutions: run.yieldedExecutions,
- });
-
- if (!response) {
- return await this.#failRunExecutionWithRetry(execution, {
- message: `Connection could not be established to the endpoint (${run.endpoint.url})`,
- });
- }
-
- const rawBody = await response.text();
-
- if (!response.ok) {
- logger.debug("Endpoint responded with non-200 status code", {
- status: response.status,
- runId: run.id,
- endpoint: run.endpoint.url,
- });
-
- const errorBody = safeJsonZodParse(errorParser, rawBody);
-
- 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, execution, errorBody.data);
- } else {
- return await this.#failRunExecutionWithRetry(execution, errorBody.data);
- }
- }
-
- // Only retry if the error isn't a 4xx
- if (response.status >= 400 && response.status <= 499) {
- return await this.#failRunExecution(this.#prismaClient, execution, {
- message: `Endpoint responded with ${response.status} status code`,
- });
- } else {
- return await this.#failRunExecutionWithRetry(execution, {
- message: `Endpoint responded with ${response.status} status code`,
- });
- }
- }
-
- const safeBody = safeJsonZodParse(parser, rawBody);
-
- if (!safeBody) {
- return await this.#failRunExecution(this.#prismaClient, execution, {
- message: "Endpoint responded with invalid JSON",
- });
- }
-
- if (!safeBody.success) {
- return await this.#failRunExecution(this.#prismaClient, execution, {
- message: generateErrorMessage(safeBody.error.issues),
- });
- }
-
- const status = safeBody.data.status;
-
- switch (status) {
- case "SUCCESS": {
- await this.#completeRunWithSuccess(execution, safeBody.data);
-
- break;
- }
- case "RESUME_WITH_TASK": {
- await this.#resumeRunWithTask(execution, safeBody.data);
-
- break;
- }
- case "ERROR": {
- await this.#failRunWithError(execution, safeBody.data);
-
- break;
- }
- case "RETRY_WITH_TASK": {
- await this.#retryRunWithTask(execution, safeBody.data);
-
- break;
- }
- case "CANCELED": {
- await this.#cancelExecution(execution);
- break;
- }
- case "UNRESOLVED_AUTH_ERROR": {
- await this.#failRunWithUnresolvedAuthError(execution, safeBody.data);
-
- break;
- }
- case "INVALID_PAYLOAD": {
- await this.#failRunWithInvalidPayloadError(execution, safeBody.data);
-
- break;
- }
- case "YIELD_EXECUTION": {
- await this.#resumeYieldedExecution(execution, safeBody.data.key);
-
- break;
- }
- default: {
- const _exhaustiveCheck: never = status;
- throw new Error(`Non-exhaustive match for value: ${status}`);
- }
- }
- }
-
- async #completeRunWithSuccess(execution: FoundRunExecution, data: RunJobSuccess) {
- const { run } = execution;
-
- return await $transaction(this.#prismaClient, async (tx) => {
- await tx.jobRun.update({
- where: { id: run.id },
- data: {
- completedAt: new Date(),
- status: "SUCCESS",
- output: data.output ?? undefined,
- queue: {
- update: {
- jobCount: {
- decrement: 1,
- },
- },
- },
- },
- });
-
- await tx.jobRunExecution.update({
- where: {
- id: execution.id,
- },
- data: {
- status: "SUCCESS",
- completedAt: new Date(),
- },
- });
- });
- }
-
- async #resumeYieldedExecution(execution: FoundRunExecution, key: string) {
- const { run } = execution;
-
- return await $transaction(this.#prismaClient, async (tx) => {
- await tx.jobRunExecution.update({
- where: {
- id: execution.id,
- },
- data: {
- status: "SUCCESS",
- completedAt: new Date(),
- run: {
- update: {
- yieldedExecutions: {
- push: key,
- },
- },
- },
- },
- });
-
- const newJobExecution = await tx.jobRunExecution.create({
- data: {
- runId: run.id,
- reason: "EXECUTE_JOB",
- status: "PENDING",
- retryLimit: EXECUTE_JOB_RETRY_LIMIT,
- },
- });
-
- await enqueueRunExecutionV1(newJobExecution, run.queue.id, run.queue.maxJobs, tx);
- });
- }
-
- async #resumeRunWithTask(execution: FoundRunExecution, data: RunJobResumeWithTask) {
- const { run } = execution;
-
- return await $transaction(this.#prismaClient, async (tx) => {
- await tx.jobRunExecution.update({
- where: {
- id: execution.id,
- },
- data: {
- status: "SUCCESS",
- completedAt: new Date(),
- },
- });
-
- // If the task has an operation, then the next performRunExecution will occur
- // when that operation has finished
- // Tasks with callbacks enabled will also get processed separately, i.e. when
- // they time out, or on valid requests to their callbackUrl
- if (!data.task.operation && !data.task.callbackUrl) {
- const newJobExecution = await tx.jobRunExecution.create({
- data: {
- runId: run.id,
- reason: "EXECUTE_JOB",
- status: "PENDING",
- retryLimit: EXECUTE_JOB_RETRY_LIMIT,
- resumeTaskId: data.task.id,
- },
- });
-
- await enqueueRunExecutionV1(
- newJobExecution,
- run.queue.id,
- run.queue.maxJobs,
- tx,
- data.task.delayUntil ?? undefined
- );
- }
- });
- }
-
- async #failRunWithError(execution: FoundRunExecution, data: RunJobError) {
- return await $transaction(this.#prismaClient, async (tx) => {
- if (data.task) {
- await tx.task.update({
- where: {
- id: data.task.id,
- },
- data: {
- status: "ERRORED",
- completedAt: new Date(),
- output: data.error ?? undefined,
- },
- });
- }
-
- await this.#failRunExecution(tx, execution, data.error ?? undefined);
- });
- }
-
- async #failRunWithUnresolvedAuthError(
- execution: FoundRunExecution,
- data: RunJobUnresolvedAuthError
- ) {
- return await $transaction(this.#prismaClient, async (tx) => {
- await this.#failRunExecution(tx, execution, data.issues, "UNRESOLVED_AUTH");
- });
- }
-
- async #failRunWithInvalidPayloadError(
- execution: FoundRunExecution,
- data: RunJobInvalidPayloadError
- ) {
- return await $transaction(this.#prismaClient, async (tx) => {
- await this.#failRunExecution(tx, execution, data.errors, "INVALID_PAYLOAD");
- });
- }
-
- async #retryRunWithTask(execution: FoundRunExecution, data: RunJobRetryWithTask) {
- const { run } = execution;
-
- return await $transaction(this.#prismaClient, async (tx) => {
- // We need to check for an existing task attempt
- const existingAttempt = await tx.taskAttempt.findFirst({
- where: {
- taskId: data.task.id,
- status: "PENDING",
- },
- orderBy: {
- number: "desc",
- },
- });
-
- if (existingAttempt) {
- await tx.taskAttempt.update({
- where: {
- id: existingAttempt.id,
- },
- data: {
- status: "ERRORED",
- error: formatError(data.error),
- },
- });
- }
-
- // We need to create a new task attempt
- await tx.taskAttempt.create({
- data: {
- taskId: data.task.id,
- number: existingAttempt ? existingAttempt.number + 1 : 1,
- status: "PENDING",
- runAt: data.retryAt,
- },
- });
-
- await tx.task.update({
- where: {
- id: data.task.id,
- },
- data: {
- status: "WAITING",
- },
- });
-
- // Now we need to create a new job execution
- const newJobExecution = await tx.jobRunExecution.create({
- data: {
- runId: run.id,
- reason: "EXECUTE_JOB",
- status: "PENDING",
- retryLimit: EXECUTE_JOB_RETRY_LIMIT,
- resumeTaskId: data.task.id,
- },
- });
-
- await enqueueRunExecutionV1(
- newJobExecution,
- run.queue.id,
- run.queue.maxJobs,
- tx,
- data.retryAt
- );
- });
- }
-
- async #failRunExecutionWithRetry(
- execution: FoundRunExecution,
- output: Record
- ): Promise {
- await $transaction(this.#prismaClient, async (tx) => {
- if (execution.retryCount + 1 > execution.retryLimit) {
- // We've reached the retry limit, so we need to fail the execution and stop retrying
- return await this.#failRunExecution(tx, execution, output);
- }
-
- // We need to retry execution
- const retryCount = execution.retryCount + 1;
- // Use an exponential backoff strategy with the exponent being 1.5
- // So when retryCount is 1, retryDelayInMs is 500ms
- // When retryCount is 2, retryDelayInMs is 750ms
- // When retryCount is 3, retryDelayInMs is 1125ms
- // When retryCount is 4, retryDelayInMs is 1687ms
- // When retryCount is 5, retryDelayInMs is 2531ms
- // When retryCount is 6, retryDelayInMs is 3796ms
- // When retryCount is 7, retryDelayInMs is 5694ms
- // When retryCount is 8, retryDelayInMs is 8541ms
- // When retryCount is 9, retryDelayInMs is 12812ms
- // When retryCount is 10, retryDelayInMs is 19218ms
- const retryDelayInMs = Math.round(500 * Math.pow(1.5, retryCount - 1));
-
- await tx.jobRunExecution.update({
- where: {
- id: execution.id,
- },
- data: {
- retryCount,
- retryDelayInMs,
- error: JSON.stringify(output),
- },
- });
-
- const runAt = new Date(Date.now() + retryDelayInMs);
-
- await enqueueRunExecutionV1(
- execution,
- execution.run.queue.id,
- execution.run.queue.maxJobs,
- tx,
- runAt
- );
- });
- }
-
- async #failRunExecution(
- prisma: PrismaClientOrTransaction,
- execution: FoundRunExecution,
- output: Record,
- status: "FAILURE" | "ABORTED" | "UNRESOLVED_AUTH" | "INVALID_PAYLOAD" = "FAILURE"
- ): Promise {
- const { run } = execution;
-
- await $transaction(prisma, async (tx) => {
- switch (execution.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,
- queue: {
- update: {
- jobCount: {
- decrement: 1,
- },
- },
- },
- },
- });
-
- 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: {
- completedAt: new Date(),
- status,
- output,
- queue: {
- update: {
- jobCount: {
- decrement: 1,
- },
- },
- },
- },
- });
-
- break;
- }
-
- await tx.jobRun.update({
- where: {
- id: run.id,
- },
- data: {
- status: "STARTED",
- startedAt: new Date(),
- },
- });
-
- const runExecution = await tx.jobRunExecution.create({
- data: {
- runId: run.id,
- reason: "EXECUTE_JOB",
- status: "PENDING",
- retryLimit: EXECUTE_JOB_RETRY_LIMIT,
- },
- });
-
- await enqueueRunExecutionV1(runExecution, run.queue.id, run.queue.maxJobs, tx);
-
- break;
- }
- }
-
- await tx.jobRunExecution.update({
- where: {
- id: execution.id,
- },
- data: {
- status: "FAILURE",
- completedAt: new Date(),
- error: JSON.stringify(output),
- },
- });
- });
- }
-
- async #cancelExecution(execution: FoundRunExecution) {
- await this.#prismaClient.jobRunExecution.update({
- where: {
- id: execution.id,
- },
- data: {
- status: "FAILURE",
- completedAt: new Date(),
- error: "This never ran because it was canceled by the user.",
- },
- });
- }
-}
-
-async function findRunExecution(prisma: PrismaClientOrTransaction, id: string) {
- return await prisma.jobRunExecution.findUnique({
- where: { id },
- include: {
- run: {
- include: {
- environment: true,
- endpoint: true,
- organization: true,
- externalAccount: true,
- queue: true,
- runConnections: {
- include: {
- integration: true,
- connection: {
- include: {
- dataReference: true,
- },
- },
- },
- },
- tasks: {
- where: {
- status: {
- in: ["COMPLETED"],
- },
- },
- },
- event: true,
- version: {
- include: {
- job: true,
- organization: true,
- },
- },
- },
- },
- },
- });
-}
diff --git a/apps/webapp/app/services/runs/performRunExecutionV2.server.ts b/apps/webapp/app/services/runs/performRunExecutionV2.server.ts
index f22d13edb..ec9318181 100644
--- a/apps/webapp/app/services/runs/performRunExecutionV2.server.ts
+++ b/apps/webapp/app/services/runs/performRunExecutionV2.server.ts
@@ -3,6 +3,7 @@ import {
BloomFilter,
ConnectionAuth,
EndpointHeadersSchema,
+ RunJobAutoYieldWithCompletedTaskExecutionError,
RunJobError,
RunJobInvalidPayloadError,
RunJobResumeWithTask,
@@ -24,9 +25,17 @@ import { safeJsonZodParse } from "~/utils/json";
import { EndpointApi } from "../endpointApi.server";
import { logger } from "../logger.server";
import { prepareTasksForCaching, prepareTasksForCachingLegacy } from "~/models/task.server";
-import { MAX_RUN_YIELDED_EXECUTIONS } from "~/consts";
+import {
+ MAX_RUN_CHUNK_EXECUTION_LIMIT,
+ MAX_RUN_YIELDED_EXECUTIONS,
+ RESPONSE_TIMEOUT_STATUS_CODES,
+ RUN_CHUNK_EXECUTION_BUFFER,
+} from "~/consts";
import { ApiEventLog } from "@trigger.dev/core";
import { RunJobBody } from "@trigger.dev/core";
+import { CompleteRunTaskService } from "~/routes/api.v1.runs.$runId.tasks.$id.complete";
+import { detectResponseIsTimeout } from "~/models/endpoint.server";
+import { forceYieldCoordinator } from "./forceYieldCoordinator.server";
type FoundRun = NonNullable>>;
type FoundTask = FoundRun["tasks"][number];
@@ -148,6 +157,7 @@ export class PerformRunExecutionV2Service {
status: "STARTED",
startedAt: new Date(),
properties: safeBody.data.properties,
+ forceYieldImmediately: false,
},
});
@@ -158,257 +168,291 @@ export class PerformRunExecutionV2Service {
}
}
async #executeJob(run: FoundRun, input: PerformRunExecutionV2Input) {
- const { isRetry, resumeTaskId } = input;
-
- if (run.status === "CANCELED") {
- await this.#cancelExecution(run);
- 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(),
- },
- });
+ const { isRetry, resumeTaskId } = input;
+ if (run.status === "CANCELED") {
+ await this.#cancelExecution(run);
return;
}
- } catch (e) {}
- const client = new EndpointApi(run.environment.apiKey, run.endpoint.url);
- const event = eventRecordToApiJson(run.event);
+ 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,
+ });
- const startedAt = new Date();
+ await this.#prismaClient.jobRun.update({
+ where: {
+ id: run.id,
+ },
+ data: {
+ status: "CANCELED",
+ completedAt: 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,
+ 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,
},
- },
- select: {
- executionCount: true,
- },
- });
-
- const connections = await resolveRunConnections(run.runConnections);
-
- if (!connections.success) {
- return this.#failRunExecution(this.#prismaClient, "EXECUTE_JOB", 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,
+ data: {
+ status: run.status === "QUEUED" ? "STARTED" : run.status,
+ startedAt: run.startedAt ?? new Date(),
+ executionCount: {
+ increment: 1,
},
- })) ?? undefined;
+ },
+ select: {
+ executionCount: true,
+ },
+ });
- if (resumedTask) {
- resumedTask = await this.#prismaClient.task.update({
+ const connections = await resolveRunConnections(run.runConnections);
+
+ if (!connections.success) {
+ return this.#failRunExecution(this.#prismaClient, "EXECUTE_JOB", 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),
+ startedAt,
+ isRetry,
+ connections.auth,
+ event,
+ sourceContext.success ? sourceContext.data : undefined
+ );
+
+ forceYieldCoordinator.registerRun(run.id);
+
+ const { response, parser, errorParser, durationInMs } = await client.executeJobRequest(
+ executionBody
+ );
+
+ forceYieldCoordinator.deregisterRun(run.id);
+
+ if (!response) {
+ return await this.#failRunExecutionWithRetry({
+ message: `Connection could not be established to the endpoint (${run.endpoint.url})`,
+ });
+ }
+
+ // Update the endpoint version if it has changed
+ const rawHeaders = Object.fromEntries(response.headers.entries());
+ const headers = EndpointHeadersSchema.safeParse(rawHeaders);
+
+ if (
+ headers.success &&
+ headers.data["trigger-version"] &&
+ headers.data["trigger-version"] !== run.endpoint.version
+ ) {
+ await this.#prismaClient.endpoint.update({
where: {
- id: resumeTaskId,
+ id: run.endpoint.id,
},
data: {
- status: resumedTask.noop ? "COMPLETED" : "RUNNING",
- completedAt: resumedTask.noop ? new Date() : undefined,
+ version: headers.data["trigger-version"],
},
});
}
- }
- const sourceContext = RunSourceContextSchema.safeParse(run.event.sourceContext);
+ const rawBody = await response.text();
- const executionBody = await this.#createExecutionBody(
- run,
- [run.tasks, resumedTask].flat().filter(Boolean),
- startedAt,
- isRetry,
- connections.auth,
- event,
- sourceContext.success ? sourceContext.data : undefined
- );
+ if (!response.ok) {
+ logger.debug("Endpoint responded with non-200 status code", {
+ status: response.status,
+ runId: run.id,
+ endpoint: run.endpoint.url,
+ });
- const { response, parser, errorParser, durationInMs } = await client.executeJobRequest(
- executionBody
- );
+ const errorBody = safeJsonZodParse(errorParser, rawBody);
- if (!response) {
- return await this.#failRunExecutionWithRetry({
- message: `Connection could not be established to the endpoint (${run.endpoint.url})`,
- });
- }
+ 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
+ );
+ } else {
+ return await this.#failRunExecutionWithRetry(errorBody.data);
+ }
+ }
- // Update the endpoint version if it has changed
- const rawHeaders = Object.fromEntries(response.headers.entries());
- const headers = EndpointHeadersSchema.safeParse(rawHeaders);
-
- if (
- headers.success &&
- headers.data["trigger-version"] &&
- headers.data["trigger-version"] !== run.endpoint.version
- ) {
- await this.#prismaClient.endpoint.update({
- where: {
- id: run.endpoint.id,
- },
- data: {
- version: headers.data["trigger-version"],
- },
- });
- }
-
- const rawBody = await response.text();
-
- if (!response.ok) {
- logger.debug("Endpoint responded with non-200 status code", {
- status: response.status,
- runId: run.id,
- endpoint: run.endpoint.url,
- });
-
- const errorBody = safeJsonZodParse(errorParser, rawBody);
-
- if (errorBody && errorBody.success) {
// Only retry if the error isn't a 4xx
- if (response.status >= 400 && response.status <= 499) {
+ if (response.status >= 400 && response.status <= 499 && response.status !== 408) {
return await this.#failRunExecution(
this.#prismaClient,
"EXECUTE_JOB",
run,
- errorBody.data
+ {
+ message: `Endpoint responded with ${response.status} status code`,
+ },
+ "FAILURE",
+ durationInMs
);
} else {
- return await this.#failRunExecutionWithRetry(errorBody.data);
+ // If the error is a timeout, we should mark this execution as succeeded (by not throwing an error) and enqueue a new execution
+ if (detectResponseIsTimeout(response)) {
+ return await this.#resumeRunExecutionAfterTimeout(
+ this.#prismaClient,
+ run,
+ input,
+ durationInMs,
+ executionCount
+ );
+ } else {
+ return await this.#failRunExecutionWithRetry({
+ message: `Endpoint responded with ${response.status} status code`,
+ });
+ }
}
}
- // Only retry if the error isn't a 4xx
- if (response.status >= 400 && response.status <= 499 && response.status !== 408) {
+ const safeBody = safeJsonZodParse(parser, rawBody);
+
+ if (!safeBody) {
return await this.#failRunExecution(
this.#prismaClient,
"EXECUTE_JOB",
run,
{
- message: `Endpoint responded with ${response.status} status code`,
+ message: "Endpoint responded with invalid JSON",
},
"FAILURE",
durationInMs
);
- } else {
- // If the error is a 504 timeout, we should mark this execution as succeeded (by not throwing an error) and enqueue a new execution
- if (response.status === 504) {
- return await this.#resumeRunExecutionAfterTimeout(
- this.#prismaClient,
+ }
+
+ if (!safeBody.success) {
+ return await this.#failRunExecution(
+ this.#prismaClient,
+ "EXECUTE_JOB",
+ run,
+ {
+ message: generateErrorMessage(safeBody.error.issues),
+ },
+ "FAILURE",
+ durationInMs
+ );
+ }
+
+ const status = safeBody.data.status;
+
+ switch (status) {
+ case "SUCCESS": {
+ await this.#completeRunWithSuccess(run, safeBody.data, durationInMs);
+
+ break;
+ }
+ case "RESUME_WITH_TASK": {
+ await this.#resumeRunWithTask(run, safeBody.data, isRetry, durationInMs, executionCount);
+
+ break;
+ }
+ case "ERROR": {
+ await this.#failRunWithError(run, safeBody.data, durationInMs);
+
+ break;
+ }
+ case "RETRY_WITH_TASK": {
+ await this.#retryRunWithTask(run, safeBody.data, isRetry, durationInMs, executionCount);
+
+ break;
+ }
+ case "CANCELED": {
+ await this.#cancelExecution(run);
+ break;
+ }
+ case "UNRESOLVED_AUTH_ERROR": {
+ await this.#failRunWithUnresolvedAuthError(run, safeBody.data, durationInMs);
+
+ break;
+ }
+ case "INVALID_PAYLOAD": {
+ await this.#failRunWithInvalidPayloadError(run, safeBody.data, durationInMs);
+
+ break;
+ }
+ case "YIELD_EXECUTION": {
+ await this.#resumeYieldedRun(
run,
- input,
+ safeBody.data.key,
+ isRetry,
durationInMs,
executionCount
);
- } else {
- return await this.#failRunExecutionWithRetry({
- message: `Endpoint responded with ${response.status} status code`,
- });
+ break;
+ }
+ case "AUTO_YIELD_EXECUTION": {
+ await this.#resumeAutoYieldedRun(
+ run,
+ safeBody.data,
+ isRetry,
+ durationInMs,
+ executionCount
+ );
+ break;
+ }
+ case "AUTO_YIELD_EXECUTION_WITH_COMPLETED_TASK": {
+ await this.#resumeAutoYieldedRunWithCompletedTask(
+ run,
+ safeBody.data,
+ isRetry,
+ durationInMs,
+ executionCount
+ );
+ break;
+ }
+ default: {
+ const _exhaustiveCheck: never = status;
+ throw new Error(`Non-exhaustive match for value: ${status}`);
}
}
- }
-
- const safeBody = safeJsonZodParse(parser, rawBody);
-
- if (!safeBody) {
- return await this.#failRunExecution(
- this.#prismaClient,
- "EXECUTE_JOB",
- run,
- {
- message: "Endpoint responded with invalid JSON",
- },
- "FAILURE",
- durationInMs
- );
- }
-
- if (!safeBody.success) {
- return await this.#failRunExecution(
- this.#prismaClient,
- "EXECUTE_JOB",
- run,
- {
- message: generateErrorMessage(safeBody.error.issues),
- },
- "FAILURE",
- durationInMs
- );
- }
-
- const status = safeBody.data.status;
-
- switch (status) {
- case "SUCCESS": {
- await this.#completeRunWithSuccess(run, safeBody.data, durationInMs);
-
- break;
- }
- case "RESUME_WITH_TASK": {
- await this.#resumeRunWithTask(run, safeBody.data, isRetry, durationInMs, executionCount);
-
- break;
- }
- case "ERROR": {
- await this.#failRunWithError(run, safeBody.data, durationInMs);
-
- break;
- }
- case "RETRY_WITH_TASK": {
- await this.#retryRunWithTask(run, safeBody.data, isRetry, durationInMs, executionCount);
-
- break;
- }
- case "CANCELED": {
- await this.#cancelExecution(run);
- break;
- }
- case "UNRESOLVED_AUTH_ERROR": {
- await this.#failRunWithUnresolvedAuthError(run, safeBody.data, durationInMs);
-
- break;
- }
- case "INVALID_PAYLOAD": {
- await this.#failRunWithInvalidPayloadError(run, safeBody.data, durationInMs);
-
- break;
- }
- case "YIELD_EXECUTION": {
- await this.#resumeYieldedRun(run, safeBody.data.key, isRetry, durationInMs, executionCount);
- break;
- }
- default: {
- const _exhaustiveCheck: never = status;
- throw new Error(`Non-exhaustive match for value: ${status}`);
- }
+ } finally {
+ forceYieldCoordinator.deregisterRun(run.id);
}
}
@@ -458,6 +502,13 @@ export class PerformRunExecutionV2Service {
cachedTaskCursor: preparedTasks.cursor,
noopTasksSet: prepareNoOpTasksBloomFilter(tasks),
yieldedExecutions: run.yieldedExecutions,
+ runChunkExecutionLimit: run.endpoint.runChunkExecutionLimit - RUN_CHUNK_EXECUTION_BUFFER,
+ autoYieldConfig: {
+ startTaskThreshold: run.endpoint.startTaskThreshold,
+ beforeExecuteTaskThreshold: run.endpoint.beforeExecuteTaskThreshold,
+ beforeCompleteTaskThreshold: run.endpoint.beforeCompleteTaskThreshold,
+ afterCompleteTaskThreshold: run.endpoint.afterCompleteTaskThreshold,
+ },
};
}
@@ -639,6 +690,7 @@ export class PerformRunExecutionV2Service {
yieldedExecutions: {
push: key,
},
+ forceYieldImmediately: false,
},
select: {
yieldedExecutions: true,
@@ -654,6 +706,101 @@ export class PerformRunExecutionV2Service {
});
}
+ async #resumeAutoYieldedRun(
+ run: FoundRun,
+ data: { location: string; timeRemaining: number; timeElapsed: number; limit?: number },
+ isRetry: boolean,
+ durationInMs: number,
+ executionCount: number
+ ) {
+ await $transaction(this.#prismaClient, async (tx) => {
+ await tx.jobRun.update({
+ where: {
+ id: run.id,
+ },
+ data: {
+ executionDuration: {
+ increment: durationInMs,
+ },
+ executionCount: {
+ increment: 1,
+ },
+ autoYieldExecution: {
+ create: [
+ {
+ location: data.location,
+ timeRemaining: data.timeRemaining,
+ timeElapsed: data.timeElapsed,
+ limit: data.limit ?? 0,
+ },
+ ],
+ },
+ forceYieldImmediately: false,
+ },
+ select: {
+ executionCount: true,
+ },
+ });
+
+ await enqueueRunExecutionV2(run, tx, {
+ isRetry,
+ skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT,
+ executionCount,
+ });
+ });
+ }
+
+ async #resumeAutoYieldedRunWithCompletedTask(
+ run: FoundRun,
+ data: RunJobAutoYieldWithCompletedTaskExecutionError,
+ isRetry: boolean,
+ durationInMs: number,
+ executionCount: number
+ ) {
+ await $transaction(this.#prismaClient, async (tx) => {
+ await tx.jobRun.update({
+ where: {
+ id: run.id,
+ },
+ data: {
+ executionDuration: {
+ increment: durationInMs,
+ },
+ executionCount: {
+ increment: 1,
+ },
+ autoYieldExecution: {
+ create: [
+ {
+ location: data.data.location,
+ timeRemaining: data.data.timeRemaining,
+ timeElapsed: data.data.timeElapsed,
+ limit: data.data.limit ?? 0,
+ },
+ ],
+ },
+ forceYieldImmediately: false,
+ },
+ select: {
+ executionCount: true,
+ },
+ });
+
+ const service = new CompleteRunTaskService(tx);
+
+ await service.call(run.environment, run.id, data.id, {
+ properties: data.properties,
+ output: data.output,
+ });
+
+ await enqueueRunExecutionV2(run, tx, {
+ isRetry,
+ skipRetrying: run.environment.type === RuntimeEnvironmentType.DEVELOPMENT,
+ executionCount,
+ });
+ });
+ }
+
async #retryRunWithTask(
run: FoundRun,
data: RunJobRetryWithTask,
@@ -748,6 +895,54 @@ export class PerformRunExecutionV2Service {
return;
}
+ const runWithLatestTask = await tx.jobRun.findUniqueOrThrow({
+ where: {
+ id: run.id,
+ },
+ select: {
+ tasks: {
+ select: {
+ id: true,
+ name: true,
+ status: true,
+ displayKey: true,
+ },
+ take: 1,
+ orderBy: { createdAt: "desc" },
+ },
+ _count: {
+ select: {
+ tasks: true,
+ },
+ },
+ },
+ });
+
+ if (runWithLatestTask._count.tasks === run._count.tasks) {
+ const latestTask = runWithLatestTask.tasks[0];
+
+ const cause =
+ latestTask?.status === "RUNNING"
+ ? `This is likely caused by task "${
+ latestTask.displayKey ?? latestTask.name
+ }" execution exceeding the function timeout`
+ : "This is likely caused by executing code outside of a task that exceeded the function timeout";
+
+ await this.#failRunExecution(
+ tx,
+ "EXECUTE_JOB",
+ run,
+ {
+ message: `Function timeout detected in ${
+ durationInMs / 1000.0
+ }s without any task creation. This is unexpected behavior and could lead to an infinite execution error because the run will never finish. ${cause}`,
+ },
+ "TIMED_OUT",
+ durationInMs
+ );
+ return;
+ }
+
await tx.jobRun.update({
where: {
id: run.id,
@@ -756,6 +951,16 @@ export class PerformRunExecutionV2Service {
executionDuration: {
increment: durationInMs,
},
+ endpoint: {
+ update: {
+ // Never allow the execution limit to be less than 10 seconds or more than MAX_RUN_CHUNK_EXECUTION_LIMIT
+ runChunkExecutionLimit: Math.min(
+ Math.max(durationInMs, 10000),
+ MAX_RUN_CHUNK_EXECUTION_LIMIT
+ ),
+ },
+ },
+ forceYieldImmediately: false,
},
});
@@ -794,6 +999,20 @@ export class PerformRunExecutionV2Service {
executionDuration: {
increment: durationInMs,
},
+ tasks: {
+ updateMany: {
+ where: {
+ status: {
+ in: ["WAITING", "RUNNING", "PENDING"],
+ },
+ },
+ data: {
+ status: status === "TIMED_OUT" ? "CANCELED" : "ERRORED",
+ completedAt: new Date(),
+ },
+ },
+ },
+ forceYieldImmediately: false,
},
});
@@ -855,7 +1074,12 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) {
return await prisma.jobRun.findUnique({
where: { id },
include: {
- environment: true,
+ environment: {
+ include: {
+ project: true,
+ organization: true,
+ },
+ },
endpoint: true,
organization: true,
externalAccount: true,
@@ -894,6 +1118,11 @@ async function findRun(prisma: PrismaClientOrTransaction, id: string) {
organization: true,
},
},
+ _count: {
+ select: {
+ tasks: true,
+ },
+ },
},
});
}
diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts
index cb9b4d6b0..15f2534a2 100644
--- a/apps/webapp/app/services/worker.server.ts
+++ b/apps/webapp/app/services/worker.server.ts
@@ -13,7 +13,6 @@ import { InvokeDispatcherService } from "./events/invokeDispatcher.server";
import { integrationAuthRepository } from "./externalApis/integrationAuthRepository.server";
import { IntegrationConnectionCreatedService } from "./externalApis/integrationConnectionCreated.server";
import { MissingConnectionCreatedService } from "./runs/missingConnectionCreated.server";
-import { PerformRunExecutionV1Service } from "./runs/performRunExecutionV1.server";
import { PerformRunExecutionV2Service } from "./runs/performRunExecutionV2.server";
import { StartRunService } from "./runs/startRun.server";
import { DeliverScheduledEventService } from "./schedules/deliverScheduledEvent.server";
@@ -21,6 +20,7 @@ import { ActivateSourceService } from "./sources/activateSource.server";
import { DeliverHttpSourceRequestService } from "./sources/deliverHttpSourceRequest.server";
import { PerformTaskOperationService } from "./tasks/performTaskOperation.server";
import { ProcessCallbackTimeoutService } from "./tasks/processCallbackTimeout";
+import { ProbeEndpointService } from "./endpoints/probeEndpoint.server";
const workerCatalog = {
indexEndpoint: z.object({
@@ -75,15 +75,15 @@ const workerCatalog = {
connectionCreated: z.object({
id: z.string(),
}),
+ probeEndpoint: z.object({
+ id: z.string(),
+ }),
simulate: z.object({
seconds: z.number(),
}),
};
const executionWorkerCatalog = {
- performRunExecution: z.object({
- id: z.string(),
- }),
performRunExecutionV2: z.object({
id: z.string(),
reason: z.enum(["EXECUTE_JOB", "PREPROCESS"]),
@@ -316,6 +316,15 @@ function getWorkerQueue() {
});
},
},
+ probeEndpoint: {
+ priority: 10,
+ maxAttempts: 1,
+ handler: async (payload, job) => {
+ const service = new ProbeEndpointService();
+
+ await service.call(payload.id);
+ },
+ },
simulate: {
maxAttempts: 5,
handler: async (payload, job) => {
@@ -341,17 +350,6 @@ function getExecutionWorkerQueue() {
shutdownTimeoutInMs: env.GRACEFUL_SHUTDOWN_TIMEOUT,
schema: executionWorkerCatalog,
tasks: {
- performRunExecution: {
- priority: 0, // smaller number = higher priority
- maxAttempts: 1,
- handler: async (payload, job) => {
- // This is a legacy task that we don't use anymore, but needs to be here for backwards compatibility
- // TODO: remove this once all performRunExecution tasks have been processed
- const service = new PerformRunExecutionV1Service();
-
- await service.call(payload.id);
- },
- },
performRunExecutionV2: {
priority: 0, // smaller number = higher priority
maxAttempts: 12,
diff --git a/apps/webapp/package.json b/apps/webapp/package.json
index b9fcc6f95..675b49e86 100644
--- a/apps/webapp/package.json
+++ b/apps/webapp/package.json
@@ -40,7 +40,6 @@
"@codemirror/view": "^6.5.0",
"@conform-to/react": "^0.6.1",
"@conform-to/zod": "^0.6.1",
- "@godaddy/terminus": "^4.12.1",
"@headlessui/react": "^1.7.8",
"@heroicons/react": "^2.0.12",
"@highlight-run/node": "^3.1.0",
diff --git a/apps/webapp/server.ts b/apps/webapp/server.ts
index e40b0a5bc..ac31b4ee2 100644
--- a/apps/webapp/server.ts
+++ b/apps/webapp/server.ts
@@ -3,7 +3,6 @@ import express from "express";
import compression from "compression";
import morgan from "morgan";
import { createRequestHandler } from "@remix-run/express";
-import { createTerminus } from "@godaddy/terminus";
const app = express();
@@ -63,23 +62,14 @@ if (process.env.HTTP_SERVER_DISABLED !== "true") {
server.keepAliveTimeout = 65 * 1000;
- // Handle shutdowns gracefully
- createTerminus(server, {
- signals: ["SIGINT", "SIGTERM"],
- timeout: process.env.GRACEFUL_SHUTDOWN_TIMEOUT
- ? Number(process.env.GRACEFUL_SHUTDOWN_TIMEOUT)
- : 5000,
- onSignal: async () => {
- console.log("[terminus] onSignal: starting cleanup");
- },
- onShutdown: async () => {
- console.log("[terminus] onShutdown: cleanup finished, server is shutting down");
- },
- onSendFailureDuringShutdown: async () => {
- console.log(
- "[terminus] onSendFailureDuringShutdown: cleanup finished, server is shutting down"
- );
- },
+ process.on("SIGTERM", () => {
+ server.close((err) => {
+ if (err) {
+ console.error("Error closing express server:", err);
+ } else {
+ console.log("Express server closed gracefully.");
+ }
+ });
});
} else {
require(BUILD_DIR);
diff --git a/docker/Dockerfile b/docker/Dockerfile
index cf12ecd6a..ce185f9ea 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -1,4 +1,4 @@
-FROM node:18.16.1-bullseye-slim AS pruner
+FROM node:18.18.2-bullseye-slim@sha256:a50436b44fe3eb2ff72696fb394ad3cae45f06cb2a7468c5ef3f012d23d9888f AS pruner
WORKDIR /triggerdotdev
@@ -7,7 +7,7 @@ RUN npx -q turbo@1.10.9 prune --scope=webapp --docker
RUN find . -name "node_modules" -type d -prune -exec rm -rf '{}' +
# Base strategy to have layer caching
-FROM node:18.16.1-bullseye-slim AS base
+FROM node:18.18.2-bullseye-slim@sha256:a50436b44fe3eb2ff72696fb394ad3cae45f06cb2a7468c5ef3f012d23d9888f AS base
RUN apt-get update && apt-get install -y openssl dumb-init
WORKDIR /triggerdotdev
COPY --chown=node:node .gitignore .gitignore
@@ -50,7 +50,7 @@ RUN pnpm run generate
RUN pnpm run build --filter=webapp...
# Runner
-FROM node:18.16.1-bullseye-slim AS runner
+FROM node:18.18.2-bullseye-slim@sha256:a50436b44fe3eb2ff72696fb394ad3cae45f06cb2a7468c5ef3f012d23d9888f AS runner
RUN apt-get update && apt-get install -y openssl
WORKDIR /triggerdotdev
RUN corepack enable
diff --git a/docker/services-compose.yml b/docker/services-compose.yml
new file mode 100644
index 000000000..b2ec2bdb7
--- /dev/null
+++ b/docker/services-compose.yml
@@ -0,0 +1,65 @@
+version: "3"
+
+volumes:
+ database-data:
+
+networks:
+ app_network:
+ external: false
+
+services:
+ db:
+ container_name: devdb
+ image: postgres:14
+ restart: always
+ volumes:
+ - database-data:/var/lib/postgresql/data/
+ environment:
+ POSTGRES_USER: postgres
+ POSTGRES_PASSWORD: postgres
+ POSTGRES_DB: postgres
+ networks:
+ - app_network
+ ports:
+ - 5432:5432
+ app:
+ build:
+ context: ../
+ dockerfile: ./docker/Dockerfile
+ ports:
+ - 3030:3030
+ depends_on:
+ - db
+ env_file:
+ - ../.env
+ environment:
+ DATABASE_URL: postgres://postgres:postgres@db:5432/postgres?schema=public
+ DIRECT_URL: postgres://postgres:postgres@db:5432/postgres?schema=public
+ SESSION_SECRET: secret123
+ MAGIC_LINK_SECRET: secret123
+ ENCRYPTION_KEY: secret123
+ REMIX_APP_PORT: 3030
+ PORT: 3030
+ WORKER_ENABLED: "false"
+ EXECUTION_WORKER_ENABLED: "false"
+ networks:
+ - app_network
+ worker:
+ build:
+ context: ../
+ dockerfile: ./docker/Dockerfile
+ depends_on:
+ - db
+ env_file:
+ - ../.env
+ environment:
+ DATABASE_URL: postgres://postgres:postgres@db:5432/postgres?schema=public
+ DIRECT_URL: postgres://postgres:postgres@db:5432/postgres?schema=public
+ SESSION_SECRET: secret123
+ MAGIC_LINK_SECRET: secret123
+ ENCRYPTION_KEY: secret123
+ REMIX_APP_PORT: 3030
+ PORT: 3030
+ HTTP_SERVER_DISABLED: "true"
+ networks:
+ - app_network
diff --git a/docs/_snippets/stable-key-param.mdx b/docs/_snippets/stable-key-param.mdx
index 1053571c7..37227f417 100644
--- a/docs/_snippets/stable-key-param.mdx
+++ b/docs/_snippets/stable-key-param.mdx
@@ -1,4 +1,4 @@
-
- Should be a stable and unique key inside the `run()`. See
+
+ Should be a stable and unique cache key inside the `run()`. See
[resumability](/documentation/concepts/resumability) for more information.
diff --git a/docs/documentation/concepts/limitations.mdx b/docs/documentation/concepts/limitations.mdx
deleted file mode 100644
index ab9b80a8c..000000000
--- a/docs/documentation/concepts/limitations.mdx
+++ /dev/null
@@ -1,31 +0,0 @@
----
-title: "Limitations"
----
-
-There are a few limitations that are important to understand.
-
-In the latest version:
-
-- Runs on localhost are limited to 5 minutes.
-- On long-running servers (not serverless) Runs can be retried erroneously.
-- Compute intensive jobs are not well supported.
-
-## Runs on localhost are limited to 5 minutes
-
-When developing locally the [CLI dev command](/documentation/guides/cli#dev-command) uses [ngrok](https://ngrok.com/) so messages can be sent to your machine.
-
-Ngrok has a timeout of 5 minutes on a Request/Response cycle. so, if a localhost Run takes longer than 5 minutes to complete, the Run will fail.
-
-This limitation will be removed in the future by adding an alternative run strategy that works well on localhost and long-running servers. This won't use the request/response cycle.
-
-## On long-running servers (not serverless) Runs can be retried erroneously
-
-Currently the only way that Runs are performed is by a Request/Response cycle when `run` is called on a Job. This is optimized for serverless functions (where you have to use a Request/Response cycle), but not for long-running servers.
-
-This limitation will be removed in the future by adding an alternative mode so Jobs works well on localhost and long-running servers. This won't use the request/response cycle.
-
-## Compute intensive jobs are not well supported
-
-Currently the only way that Runs are performed is inside a Request/Response cycle when `run` is called on a Job. This is not a good way to perform compute intensive jobs.
-
-In the future we will add good support for compute intensive jobs.
diff --git a/docs/documentation/concepts/limits.mdx b/docs/documentation/concepts/limits.mdx
new file mode 100644
index 000000000..1bb7b0d0f
--- /dev/null
+++ b/docs/documentation/concepts/limits.mdx
@@ -0,0 +1,116 @@
+---
+title: "Limits"
+---
+
+## General Limits
+
+The following limits apply to the Trigger.dev Cloud service and users of the self-hosted version of Trigger.dev.
+
+| | Hobby | Team | Self-hosted / Enterprise |
+| ----------------------------------------------------------------------- | --------- | ----------- | ------------------------- |
+| Team Members | Up to 2 | Up to 5 | Custom |
+| Projects | 1 | Up to 5 | Custom |
+| Jobs per Project | Up to 10 | Up to 50 | Custom |
+| Runs (per Month) | 5,000 | Up to 1m | Custom |
+| Run Log retention | 24 hours | 7 days | Custom |
+| Connected Integrations | Up to 50 | Up to 1000 | Custom |
+| [Tasks per Run](#tasks-per-runs) | Up to 250 | Up to 1000 | Custom |
+| [Concurrent Run Executions](#concurrent-run-executions) | Up to 10 | Up to 10 | Custom |
+| [Maximum Task Duration](#maximum-task-duration) | < 2m | < 2m | < Deployment Grace Period |
+| [Maximum Run Execution Duration](#maximum-total-run-execution-duration) | up to 15m | up to 2 hrs | Custom |
+| [Yielded Executions per Run](#yielded-executions-per-run) | Up to 100 | Up to 100 | Custom |
+
+### Tasks per Run
+
+For any individual Job Run, the number of Tasks that can be executed is limited to 250 for Hobby and 1000 for Team plans. This limit is enforced to prevent runaway Jobs from consuming excessive resources.
+
+#### What is a Task?
+
+Tasks are the fundamental building blocks on which the Trigger.dev service is constructed. You can create and run a task using [io.runTask()](/sdk/io/runtask):
+
+```ts
+client.defineJob({
+ id: "task-example",
+ name: "Task Example",
+ version: "1.0.0",
+ trigger: eventTrigger({ name: "task.example" }),
+ run: async (payload, io, ctx) => {
+ const response = await io.runTask("task-1", async (task) => {
+ // Do some work here
+ return { foo: "bar" };
+ });
+ },
+});
+```
+
+Tasks power the following features as well:
+
+- [io.wait()](/sdk/io/wait)
+- [io.sendEvent()](/sdk/io/sendevent)
+- [io.backgroundFetch()](/sdk/io/backgroundfetch)
+- [io.logger](/sdk/io/logger)
+
+Our integration clients are also built on top of Tasks, so any time you call an integration client method, you are creating a Task. e.g.:
+
+```ts
+client.defineJob({
+ id: "send-resend-email",
+ name: "Send Resend Email",
+ version: "0.1.0",
+ trigger: eventTrigger({
+ name: "send.email",
+ }),
+ integrations: {
+ resend,
+ },
+ run: async (payload, io, ctx) => {
+ // This creates a Task
+ await io.resend.sendEmail("send-email", {
+ to: payload.to,
+ subject: payload.subject,
+ text: payload.text,
+ from: "Trigger.dev ",
+ });
+ },
+});
+```
+
+Anything that shows up as an item on the [Run Log](/documentation/guides/viewing-runs#run-page) is a Task:
+
+
+
+### Concurrent Run Executions
+
+A Run Execution is a single HTTP request from the Trigger.dev server to your endpoint to execute a run. The number of concurrent Run Executions is limited to 10 for Hobby and Team plans.
+
+This does not include runs that are waiting for a [io.wait()](/sdk/io/wait) to complete, so you could in theory have 1000s of "In Progress" jobs at a given time with no current run executions.
+
+Going over this limit does not abort or cancel runs, but it will prevent new run executions until the number of concurrent executions drops below the limit.
+
+### Maximum Task Duration
+
+The Maximum Task Duration is the maximum amount of time a single Task can run for. This limit is partly enforced by the Trigger.dev server, but also by the execution runtime of your deployed serverless function.
+
+For example, if you're deploying to Vercel and using their Node.js Serverless functions, the maximum execution time is anywhere from 1 second to 5 minutes. If you have a single task that can run for longer than your maximum function execution time, it will never complete.
+
+We will retry tasks that never complete due to a timeout, but if the task continues to not complete, it will be marked as cancelled and the run will be timed out with an output like the following:
+
+```json
+{
+ "message": "Function timeout detected in 10s without any task creation. This is unexpected behavior and could lead to an infinite execution error because the run will never finish. This is likely caused by task \"initial-long-task\" execution exceeding the function timeout"
+}
+```
+
+See our Next.js section on [Deployment](/documentation/guides/platforms/nextjs#deployment) for more information on how to configure your function timeout.
+
+Additionally, the Trigger.dev enforces a soft-cap of 2 minutes. Tasks that take longer than 2 minutes will be allowed to complete but we cannot guarentee that they won't be retried erroneously or cause your run execution to be locked for up to 4 hours. This is because of a current limitation of [Graphile Worker](https://github.com/graphile/worker) and our deployment platform.
+
+### Maximum Total Run Execution Duration
+
+The Maximum Total Run Execution Duration is the maximum amount of time a single run can execute for. Runs are completed over 1 or more executions, depending on many factors like the number of tasks, the serverless function timeout, task errors and delays. The Trigger.dev measures the total time spent across all run executions and will cancel the run if it exceeds the limit.
+
+Hobby plans have a limit of 15 minutes, Team plans have a limit of 2 hours, and Enterprise plans can set a custom limit.
+
+### Yielded Executions per Run
+
+You can manually yield a run execution using `io.yield()`, which will exit the current run execution and schedule a new run execution to continue the run. You do this at most 100 times per run.
diff --git a/docs/documentation/concepts/resumability.mdx b/docs/documentation/concepts/resumability.mdx
index e6e9b5c84..675cfc0f8 100644
--- a/docs/documentation/concepts/resumability.mdx
+++ b/docs/documentation/concepts/resumability.mdx
@@ -8,14 +8,14 @@ description: "Runs are resumable by returning Task stored data"
## How does this work?
1. When a Run is created, it is given a unique ID. This ID is used to identify the Run.
-2. [Tasks](/documentation/concepts/tasks) have a `key` which is a string and is the first parameter. This should be stable and unique inside that `run` function.
+2. [Tasks](/documentation/concepts/tasks) have a `cacheKey` which is a string and is the first parameter. This should be stable and unique inside that `run` function.
3. When a Task is completed, its output is stored.
4. If a Run exceeds the timeout, or your server restarts, the Run will be "replayed".
5. The second+ time it is run, Tasks that have already successfully completed will immediately return their first output. The code inside them won't re-run.
-## How to use keys
+## How to use cache keys
-Like we mentioned above, we use Task keys to determine which Tasks have already been executed. They are defined by you inside your `run` function, for example when you call `io.slack.postMessage`:
+Like we mentioned above, we use Task cache keys to determine which Tasks have already been executed. They are defined by you inside your `run` function, for example when you call `io.slack.postMessage`:
```ts
await io.slack.postMessage("⭐️ New Star", {
@@ -24,9 +24,9 @@ await io.slack.postMessage("⭐️ New Star", {
});
```
-In this example, the key is the string `"⭐️ New Star"`. This means that if the Job is interrupted and then resumed, the `slack.postMessage` Task will be skipped because it has already been executed.
+In this example, the cache key is the string `"⭐️ New Star"`. This means that if the Job is interrupted and then resumed, the `slack.postMessage` Task will be skipped because it has already been executed.
-If you make multiple calls to `slack.postMessage`, you should use different keys for each call. For example:
+If you make multiple calls to `slack.postMessage`, you should use different cache keys for each call. For example:
```ts
await io.slack.postMessage("⭐️ New Star", {
@@ -42,7 +42,9 @@ await io.slack.postMessage("🚨 Critical Issue", {
If you are calling a Task multiple times with the same key, it will only be executed once. For example, if you call `slack.postMessage` with the key `"⭐️ New Star"` twice, it will only be executed once.
-## How to use keys with loops
+See our [Task concept guide](/documentation/concepts/tasks) for more information about tasks and how they are crucial to the resumability of your Jobs.
+
+## How to use cache keys with loops
If you are using a loop, you should use the loop index as the key. For example:
@@ -56,3 +58,8 @@ for (let i = 0; i < 10; i++) {
});
}
```
+
+
+ We don't currently support running tasks in parallel so `Promise.all` will not work correctly.
+ This is something on our roadmap that we hope to support soon.
+
diff --git a/docs/documentation/concepts/tasks.mdx b/docs/documentation/concepts/tasks.mdx
index 288872f60..31539a6d9 100644
--- a/docs/documentation/concepts/tasks.mdx
+++ b/docs/documentation/concepts/tasks.mdx
@@ -3,17 +3,24 @@ title: "Tasks"
description: "Tasks are individual building blocks of a Run."
---
-> A Task is a resumable unit of a Run that can be retried, resumed and is logged.
+A [Task](/documentation/concepts/tasks) is a cached unit of work in a Job Run that are logged to the Trigger.dev UI.
-## Tasks vs regular code
+
+ Any interaction with an external service (database or API) should be wrapped in a Task. Failing to
+ do so could result in repeated work when runs are resumed.
+
-In the `run()` function you can use regular code and you can use Tasks.
+## Why do you need tasks?
+
+Tasks are a key building block of how Trigger.dev works, and failing to use them will result in unpredictable results. Tasks allow bits of work inside a Job Run to be cached and the results of those tasks to be reused.
+
+This is **very important** because for a Job Run to be resumable (e.g. after a serverless function timeout, or because of a call to `io.wait()`), we need to call the `Job.run` function multiple times. If we didn't cache the results of Tasks, then we would be repeating work on each run.
```ts
client.defineJob({
id: "new-user",
name: "Run when a new user signs up",
- version: "0.0.1",
+ version: "1.0.0",
trigger: eventTrigger({
name: "new.user",
schema: z.object({
@@ -24,26 +31,24 @@ client.defineJob({
resend,
},
run: async (payload, io, ctx) => {
- // regular code, not a Task
- // the inputs/outputs of this function are not sent to the Trigger.dev platform
- const user = await prisma.user.findUnique({
+ // This code will run twice. Once when the run first starts, and once after the wait
+ const user = await prisma.user.findUniqueOrThrow({
where: { id: payload.userId },
select: { email: true, name: true },
});
- if (!user) throw new Error(`User not found: ${payload.userId}`);
- // Integration functions are Tasks
- await io.resend.sendEmail("Welcome email", {
+ // This code will run once, because the resend integration creates a task with the "welcome-email" cacheKey
+ await io.resend.sendEmail("welcome-email", {
to: user.email,
from: "jane@acme.inc",
subject: "Welcome!",
html: welcomeEmail(user.name),
});
- // built-in io functions are Tasks
+ // This code will run once, because io.wait creates a task with the "wait" cacheKey
await io.wait("wait", 60 * 60 * 3); // wait for 3 hours
- // You can wrap your own code in a Task, for retrying, resumability and logging
+ // This code will run once, because we're manually creating a task with the "my-task" cacheKey
const response = await io.runTask(
"my-task",
async () => {
@@ -57,21 +62,246 @@ client.defineJob({
});
```
-## The benefits of Tasks
+As well as powering the resumable nature of Trigger.dev, Tasks also provide:
-Tasks are a powerful concept that gives you a lot of benefits:
-
-- **Resumability** – Runs can exceed the maximum timeout on serverless platforms. If a Run exceeds this limit, it will be re-run. When it is re-run, any completed Tasks return their original output and they aren't re-run. Read more about [Resumability](/documentation/concepts/resumability).
-- **Retryable** – If a Task fails, it will be retried. You can configure how (or if) a Task is retried. Full details in the [io SDK reference](/sdk/io).
+- **Retryable** – If a Task fails, it can be retried. You can configure how (or if) a Task is retried. Full details in the [io.runTask() SDK reference](/sdk/io/runtask).
- **Logging** – Tasks are logged, so you can see what happened in a Run. Find out more about [viewing runs](/documentation/guides/viewing-runs).
+## Task Cache Keys
+
+The first param of all Tasks is a `cacheKey`. This is a unique identifier for the Task inside that Run. It is used for storing the cached result of a task. It is also used to identify the Task in the [Viewing Runs Dashboard](/documentation/guides/viewing-runs).
+
+It's important that cacheKey's are unique inside an individual Job Run.
+
+## Creating Tasks
+
+There are **3** ways of using tasks in your code:
+
+- Using the [io.runTask()](/sdk/io/runtask) function
+- Using one of our [Integration packages](/documentation/concepts/integrations) and calling the `io.integration.runTask()` wrapper
+- Using one of our [Integration packages](/documentation/concepts/integrations) and calling a task wrapper function, such as [io.slack.postMessage()](/integrations/apis/slack)
+
+### Using `io.runTask()`
+
+The `io.runTask()` function allows you to run a Task manually. It takes a `cacheKey` and a function to run. The function will only be run if the Task is not already cached.
+
+```ts
+const response = await io.runTask("my-task", async (task) => {
+ return await longRunningCode(payload.userId);
+});
+```
+
+The callback function is passed a `task` object, which can be useful for providing an idempotency key to an external service. For example, Stripe:
+
+
+ Our [Stripe Integration](integrations/apis/stripe) handles this for you automatically, this is
+ just for documentation purposes
+
+
+```ts
+const stripe = new Stripe(process.env.STRIPE_SECRET_KEY, {
+ apiVersion: "2020-08-27",
+});
+
+await io.runTask("create-customer", async (task) => {
+ await stripe.customers.create(
+ {
+ email: "eric@trigger.dev",
+ },
+ {
+ idempotencyKey: task.idempotencyKey,
+ }
+ );
+});
+```
+
+`runTask` also takes an optional 3rd argument, which allows you to customize how the Task is displayed and run. For example, you can supply a name and some properties to be displayed in the Viewing Runs Dashboard:
+
+```ts
+const response = await io.runTask(
+ "my-task",
+ async (task) => {
+ return await longRunningCode(payload.userId);
+ },
+ {
+ name: "My Task",
+ properties: [
+ {
+ label: "User ID",
+ value: payload.userId,
+ },
+ ],
+ icon: "user",
+ }
+);
+```
+
+See the [io.runTask() SDK reference](/sdk/io/runtask) for more information.
+
+### Using `io.integration.runTask()`
+
+All of our [Integration packages](/documentation/concepts/integrations) expose a `runTask()` function. The main differences between this and `io.runTask()` are:
+
+- Adds an additional callback parameter which provides the underlying authenticated integration client
+- Automatically sets the `icon` property on the Task.
+- Configures sensible defaults for retries and error handling.
+
+An example here demonstrates using the GitHub integration's `runTask` function to create a project card when a new user signs up:
+
+```ts
+import { Github } from "@trigger.dev/github";
+
+const github = new Github({
+ id: "github",
+});
+
+client.defineJob({
+ id: "create-project-card",
+ name: "Create Project Card",
+ version: "1.0.0",
+ trigger: eventTrigger({
+ name: "new.user",
+ }),
+ integrations: {
+ github,
+ },
+ run: async (payload, io, ctx) => {
+ await io.github.runTask(
+ "create-card",
+ async (client, task) => {
+ // client is an authenticated GitHub client (https://github.com/octokit/octokit.js)
+ return client.rest.projects.createCard({
+ column_id: process.env.GITHUB_PROJECT_COLUMN_ID,
+ note: `New User ${payload.user.name} signed up!`,
+ });
+ },
+ { name: "Create card" }
+ );
+ },
+});
+```
+
+### Using an Integration Task Wrapper Function
+
+Our [Integration packages](/documentation/concepts/integrations) also expose a number of task wrapper functions. These are functions that wrap a common task for that integration. For example, the [Slack integration](/integrations/apis/slack) exposes a `postMessage()` function:
+
+```ts
+import { Slack } from "@trigger.dev/slack";
+
+const slack = new Slack({
+ id: "slack",
+});
+
+client.defineJob({
+ id: "send-welcome-message",
+ name: "Send welcome message",
+ version: "1.0.0",
+ trigger: eventTrigger({
+ name: "new.user",
+ }),
+ integrations: {
+ slack,
+ },
+ run: async (payload, io, ctx) => {
+ await io.slack.postMessage("send-message", {
+ channel: process.env.SLACK_CHANNEL_ID,
+ text: `New user ${payload.user.name} signed up!`,
+ });
+ },
+});
+```
+
+All task wrapper functions take a `cacheKey` as the first argument, because they are Tasks under the hood. Think of them as a convenience wrapper around `io.runTask()`.
+
+We strive to document all of the task wrapper functions in our [Integration packages](/documentation/concepts/integrations). For example, checkout our [GitHub integration task](/integrations/apis/github-tasks) docs.
+
## Subtasks
-A Task can have multiple subtasks, and so on. This is useful for breaking down a large Task into smaller Tasks. We currently support nesting 5 levels deep.
+You can break up a task into multiple subtasks. This is useful for breaking up a long-running task into smaller chunks, while consolidating the logging into a single task in the dashboard with children.
-## Task Keys
+We currently support nesting up to 5 levels
-The first param of all Tasks is a `key`. This is a unique identifier for the Task inside that Run. It is used for resumability and logging. It is also used to identify the Task in the [Viewing Runs Dashboard](/documentation/guides/viewing-runs).
+```ts
+const response = await io.runTask("parent-task", async (task) => {
+ await io.runTask("child-1", async () => {
+ // do something
+ });
+
+ await io.runTask("child-2", async () => {
+ // do something
+ });
+});
+```
+
+Task cacheKey's are automatically scoped to the parent task. So for example, you can reuse a cacheKey inside a parent task and it will not conflict with another top-level task.
+
+```ts
+const response = await io.runTask("parent-task", async (task) => {
+ await io.runTask("child-1", async () => {
+ // do something
+ });
+
+ await io.runTask("child-2", async () => {
+ // do something
+ });
+});
+
+// This will not conflict with the child-1 task above
+const response = await io.runTask("child-1", async (task) => {
+ // do something
+});
+```
+
+### Extracting Common Tasks
+
+Subtasks allow you to DRY up any repeating task code into a single function. For example, if you have a common task that sends a welcome email, you can extract that into a function:
+
+```ts
+const sendWelcomeEmail = async (cacheKey: string, io: IO, resend: Resend, userId: string) => {
+ return await io.runTask(cacheKey, async () => {
+ const user = await io.runTask("fetch-user", async () => {
+ return prisma.user.findUniqueOrThrow({
+ where: { id: userId },
+ select: { email: true, name: true },
+ });
+ });
+
+ await io.resend.sendEmail("📧", {
+ to: user.email,
+ from: "eric@trigger.dev",
+ subject: "Welcome!",
+ html: welcomeEmail(user.name),
+ });
+ });
+};
+
+client.defineJob({
+ id: "new-user",
+ name: "Run when a new user signs up",
+ version: "1.0.0",
+ trigger: eventTrigger({
+ name: "new.user",
+ schema: z.object({
+ userId: z.string(),
+ }),
+ }),
+ integrations: {
+ resend,
+ },
+ run: async (payload, io, ctx) => {
+ await sendWelcomeEmail("🫡", io, io.resend, payload.userId);
+ },
+});
+```
+
+
+ Always make sure you are allow passing a unique cacheKey to the `runTask` function, so the tasks
+ inside the function are not accidentally reused.
+
+
+## Limitations
+
+A single task has an upper-bound on it's execution duration, which must be less than the serverless function execution timeout of your deployed platform. For more information see our [Limits docs](/documentation/concepts/limits#maximum-task-duration)
## References
diff --git a/docs/documentation/concepts/what-is-triggerdotdev.mdx b/docs/documentation/concepts/what-is-triggerdotdev.mdx
index 7793e5090..4be93be5b 100644
--- a/docs/documentation/concepts/what-is-triggerdotdev.mdx
+++ b/docs/documentation/concepts/what-is-triggerdotdev.mdx
@@ -103,3 +103,19 @@ Below is a simplified architecture diagram of how Trigger.dev works:
As you can see above, we communicate between your code and the Trigger.dev platform. This allows us to send events to your code, and receive tasks from your code.
+
+## Limitations
+
+There are a few limitations that are important to understand.
+
+In the latest version the following are not supported:
+
+### Long-running servers
+
+Currently Trigger.dev is optimized for deployment to serverless functions, but not for long-running servers.
+
+This limitation will be removed in the future by adding an alternative mode so Jobs works well on localhost and long-running servers.
+
+### Compute intensive tasks
+
+Because Trigger.dev is optimized for serverless functions, it is not well suited for compute intensive jobs as each individual task is limited to the [Maximum Run Chunk Execution Duration](/placeholder)
diff --git a/docs/documentation/guides/create-a-job.mdx b/docs/documentation/guides/create-a-job.mdx
deleted file mode 100644
index c89c05dfd..000000000
--- a/docs/documentation/guides/create-a-job.mdx
+++ /dev/null
@@ -1,270 +0,0 @@
----
-title: "Create a Job"
-description: "How to create a Job in your codebase"
----
-
-> Jobs are the core of the system. They allow you to run code when some event occurs. They are built using a combination of Triggers and Tasks.
-
-### Pre-requisites
-
-Make sure your Project is set up with Trigger.dev. We recommend [using the CLI](/documentation/quickstart) to do this.
-
-## How to write a Job in code
-
-### 1. Create a Job file in your Project
-
-This is where you will write your Job code. E.g. `my-job.ts`.
-
-```ts
-//this path might be different depending on your project
-import { client } from "@/trigger";
-
-client.defineJob({
- // This is the unique ID for your Job's end-point
- id: "your-job-id",
- // This is the name of your Job
- name: "Your Job name",
- // This is the version of our SDK you are using
- version: "0.0.1",
- ...
-```
-
-The `id` and `name` are important because they are used to create and identify your Job in the app.
-
-
- This Job must be imported in the `trigger` file in order to be registered when the CLI dev command
- is run. If you're using Next.js, this can be found in either the `app/api/trigger/route.ts` file
- for projects using the App Router, or `pages/api/trigger.ts` if you're using the Pages Router.
-
-
-### 2. Choose a Trigger
-
-This is what kicks-off a Job. There are a few different types of Triggers you can use:
-
-
-
- Run a Job on a repeating schedule, using [intervalTrigger](/documentation/concepts/triggers/scheduled#interval)
- ```ts
- client.defineJob({
- ...
- trigger: intervalTrigger({
- seconds: 60,
- }),
- ...
- ```
- Or with CRON syntax, using [cronTrigger](/documentation/concepts/triggers/scheduled#using-cron-syntax):
- ```ts
- client.defineJob({
- ...
- trigger: cronTrigger({
- cron: "30 14 * * 1",
- }),
- ...
- ```
-
-
-
- Start your Jobs when an event happens in another API. You'll need to use [Integrations](/integrations) to do this.
-
- Here's an example with the GitHub integration.
-
- ```ts
- client.defineJob({
- ...
- //E.g. When a GitHub issue is modified on the triggerdotdev/trigger.dev repo
- trigger: github.triggers.repo({
- event: events.onIssue,
- owner: "triggerdotdev",
- repo: "trigger.dev",
- }),
- ...
- ```
-
-
-
- The [eventTrigger](/documentation/concepts/triggers/events) allows you to define an event that your Job listens for.
-
- When you [send an event](/documentation/concepts/triggers/events#sending-events) with the same name the Job will run.
-
- ``` ts
- client.defineJob({
- ...
- //E.g. when a user is created in your app (you send the event)
- trigger: eventTrigger({
- name: "user.created",
- schema: z.object({
- name: z.string(),
- email: z.string(),
- paidPlan: z.boolean(),
- }),
- ...
- ```
-
-
-
- These are advanced features that allows you to attach dynamic triggers to a Job. Full information [here](/documentation/concepts/triggers/dynamic).
-
-
-
-### 3. Create the Job Tasks
-
-> A Task is a resumable unit of a Run that can be retried, resumed and is logged.
-
-
- You can use just regular code in your Jobs. But you don't get the benefits of retrying, logging
- and resumability. More info on [Tasks vs regular
- code](/documentation/concepts/tasks#tasks-vs-regular-code).
-
-
-You can string together multiple Tasks and regular code in any order you want.
-
-**Useful built-in Tasks:**
-
-| Task | Description | Task code |
-| ------------------------------------------- | ----------------------------------------------------------------------- | ------------------------------------------------------------------------------------ |
-| [Delay](/documentation/concepts/delays) | Wait for a period of time | `await io.wait("wait", 60);` |
-| [Log](/sdk/io/logger) | Log a message | `await io.logger.log("Hello");` |
-| [Send Event](/sdk/io/sendevent) | Send an event (for eventTrigger) | `await io.sendEvent("my-event", { name: "my.event", payload: { hello: "world" } });` |
-| [Run task](/sdk/io/runtask) | Wrap your own code in this to create a Task | `await io.runTask("My Task", async () => { console.log("Hello"); });` |
-| [Background fetch](/sdk/io/backgroundfetch) | Fetch data from a URL that can take longer that the serverless timeout. | `await io.backgroundFetch("fetch-some-data", { url: "https://example.com" });` |
-
-For a full list of built-in Tasks, see the [io SDK reference](/sdk/io).
-
-**Integration Task examples:**
-
-
- To use our integrations you will need to set them up in the app first. Our guide is
- [here](/documentation/guides/using-integrations).
-
-
-
-
-
-**Task:** [backgroundCreateCompletion](/integrations/apis/openai)
-
-```ts
-await io.openai.backgroundCreateChatCompletion("background-chat-completion", {
- model: "gpt-3.5-turbo",
- messages: [
- {
- role: "user",
- content: "Create a good programming joke about background jobs",
- },
- ],
-});
-```
-
-View more OpenAI tasks [here](/integrations/apis/openai).
-
-
-
-
-**Task:** [addIssueLabels](/integrations/apis/github-tasks)
-
-```ts
-await io.github.addIssueLabels("add label", {
- owner: payload.repository.owner.login,
- repo: payload.repository.name,
- issueNumber: payload.issue.number,
- labels: ["bug"],
-});
-```
-
-View more GitHub tasks [here](/integrations/apis/github-tasks).
-
-
-
-
-**Task:** [sendEmail](/integrations/apis/resend)
-
-```ts
-await io.resend.sendEmail("send-email", {
- to: payload.to,
- subject: payload.subject,
- text: payload.text,
- from: "Trigger.dev ",
-});
-```
-
-
-
-
-**Task:** [postMessage](/integrations/apis/slack)
-
-```ts
-await io.slack.postMessage("post message", {
- channel: "C04GWUTDC3W",
- text: "My first Slack message",
-});
-```
-
-View more Slack tasks [here](/integrations/apis/slack).
-
-
-
-
-These are just a few examples of Integration Tasks. For many more, browse our [Integrations section](/integrations/).
-
-### 4. Register your Jobs
-
-While your app is running, open a **new terminal window or tab** and run:
-
-
-
-```bash npm
-npx @trigger.dev/cli@latest dev
-```
-
-```bash pnpm
-pnpm dlx @trigger.dev/cli@latest dev
-```
-
-```bash yarn
-yarn dlx @trigger.dev/cli@latest dev
-```
-
-
-
-This will register all of your Jobs, they should appear in your dashboard.
-
-
- Not seeing your Job in the web app? It might be because you forgot to import it. This will need to
- be either in `app/api/trigger/route.ts` file if you're using the Next,js App Router, or
- `pages/api/trigger.ts` if you're using the Next,js Pages Router.
-
-
-If you are having trouble getting your job running, please reach out to us and we will help you fix any issues:
-
-- [Join our Discord](https://discord.gg/kA47vcd8P6)
-- [Email us](mailto:help@trigger.dev)
-
----
-
-## Next steps
-
-We recommend exploring all of the below sections to fully understand how to create and run Jobs using Trigger.dev.
-
-
-
- A guide for how to run your Jobs. Triggering a test Run and Triggering your Job for real
-
-
-
-
- View example Jobs / the example jobs repo. These are a great starting
- point for creating your own Jobs.
-
-
- How to use the SDK. This includes all the available Tasks, triggers and
- actions you can use.
-
-
-
- Integrations make it easy to authenticate and use APIs.
-Learn how to use and create integrations.
-
-
diff --git a/docs/documentation/guides/platforms/nextjs.mdx b/docs/documentation/guides/platforms/nextjs.mdx
index 06685f8fe..dfca5a234 100644
--- a/docs/documentation/guides/platforms/nextjs.mdx
+++ b/docs/documentation/guides/platforms/nextjs.mdx
@@ -17,6 +17,14 @@ View our [guide for writing Jobs](/documentation/guides/create-a-job).
View our [deployment guide](/documentation/guides/deployment) to learn how to deploy your Jobs.
+### Serverless function timeouts
+
+If you are deploying your Next.js app to Vercel, you may need to configure a larger max function duration. By default, Vercel has a max function duration of 10 seconds. As outlined in our [Limits docs](/documentation/concepts/limits#maximum-task-duration), the max function duration effects the maximum [Task](/documentation/concepts/tasks) duration.
+
+So if you have any tasks that may take longer than 10 seconds to run (or close to 10 seconds), you should increase the max function duration to a higher value (only available to paid Vercel plans).
+
+See the [Vercel docs](https://vercel.com/docs/functions/serverless-functions/runtimes#max-duration) for more information on increasing the `maxDuration` for your Vercel Serverless Functions.
+
## Middleware
Next.js Middleware allows you to run code before a request is completed, and if you are using it currently in your Next.js project (or you add it later), you might need to guard against altering requests to the `/api/trigger` endpoint, which needs to be exposed to the Trigger.dev installation (either your self-hosted one or the Trigger.dev Cloud).
diff --git a/docs/documentation/guides/writing-jobs-step-by-step.mdx b/docs/documentation/guides/writing-jobs-step-by-step.mdx
new file mode 100644
index 000000000..438049da0
--- /dev/null
+++ b/docs/documentation/guides/writing-jobs-step-by-step.mdx
@@ -0,0 +1,382 @@
+---
+title: "Writing Jobs - Step by Step"
+description: "Best practices for writing well-behaving Jobs in your codebase"
+---
+
+## Pre-requisites
+
+This guide assumes you already have a project setup and working with Trigger.dev. If not, head over to our [Quick Start guides](/documentation/quickstarts/introduction) to get up and running in a few minutes.
+
+## 1. Define your Job
+
+A Job is a collection of Tasks that are run in a specific order. You can think of it as a function that you can run on a schedule, or when an event happens. Jobs are defined by calling the `TriggerClient.defineJob` function
+
+```ts
+//this path might be different depending on your project
+import { client } from "@/trigger";
+
+client.defineJob({
+ // ... job definition
+});
+```
+
+
+ If you aren't seeing defined jobs in your Trigger.dev dashboard, it might be because the job file
+ isn't being imported in your app.
+
+
+## 2. Choose a name and ID
+
+Each job must have a unique and stable `id` and `name`. The `id` is used to identify the Job in the database, and the `name` is used to identify the Job in the UI.
+
+```ts
+client.defineJob({
+ id: "my-job",
+ name: "My Job",
+ // ... job definition
+});
+```
+
+We will pass the value of the `id` property through a slugifier because we use it in URLs in our Dashboard. This means you can use any characters you want, but we recommend using only lowercase letters, numbers and dashes.
+
+## 3. Set the current version
+
+The `version` property is used to track changes to your Job. It's required to be a [semantic version](https://semver.org/) string. You can track changes to your Job by incrementing the version number, and we will display in the Dashboard which version each Job Run was created with.
+
+```ts
+client.defineJob({
+ id: "my-job",
+ name: "My Job",
+ version: "1.0.0",
+ // ... job definition
+});
+```
+
+## 4. Choose a Trigger
+
+The trigger you choose determines how and when a job will run. See our [Triggers guide](/documentation/concepts/triggers/introduction) for more information. The Trigger you choose also defines the type of the run `payload` argument (more in this below)
+
+```ts
+client.defineJob({
+ id: "my-job",
+ name: "My Job",
+ version: "1.0.0",
+ trigger: eventTrigger({
+ name: "my.event",
+ }),
+ // ... job definition
+});
+```
+
+## 5. Add integrations
+
+Integrations provide a convienent way to create and run tasks against authenticated APIs inside your Job's run function. You'll need to pass them in the `integrations` option when defining your Job.
+
+```ts
+import { Slack } from "@trigger.dev/slack";
+
+const slack = new Slack({ id: "slack" });
+
+client.defineJob({
+ id: "my-job",
+ name: "My Job",
+ version: "1.0.0",
+ trigger: eventTrigger({
+ name: "my.event",
+ }),
+ integrations: { slack },
+ // ... job definition
+});
+```
+
+## 6. Implement the run function
+
+The `run` function implements your custom code that will be executed when your Job is run. It's an async function that takes three arguments:
+
+- The run `payload` - The type of the `payload` argument is determined by the Trigger you choose.
+- An instance of `IO`, which exposes built-in tasks and allows you to create your own, as well as interact with integrations.
+- A `context` object, which contains information about the current run, such as the run ID, the Job ID, and the Job version. [Context reference](/sdk/context)
+
+```ts
+import { Slack } from "@trigger.dev/slack";
+
+const slack = new Slack({ id: "slack" });
+
+client.defineJob({
+ id: "my-job",
+ name: "My Job",
+ version: "1.0.0",
+ trigger: eventTrigger({
+ name: "my.event",
+ }),
+ integrations: { slack },
+ run: async (payload, io, context) => {
+ // ... your code
+ },
+});
+```
+
+
+ We do not compile and ship your code to run on the Trigger.dev server. It runs exactly where
+ you've deployed your code (e.g. Vercel).
+
+
+The `run` function you define is like a normal JavaScript function in all respects except one: it will be called one or more times to complete a single Job Run.
+
+This means that you can't rely on any state that is not persisted between runs, and you must **create tasks** to ensure that work is not repeated.
+
+
+ The run function is called multiple times to ensure that your Job is resilient to failure and can finish running even if it is interrupted. There are many reasons why a run could be interrupted and resumed later, including:
+
+- The serverless function times out
+- A task fails and needs to be retried
+- A wait task is used to delay continuing the run until a later time
+- A run yields execution to prevent the serverless function from timing out
+- Waiting for a [backgroundFetch](/sdk/io/backgroundfetch) to complete
+- Waiting for a task callback to be called (like the ones used in our [Replicate integration](/integrations/apis/replicate#predictions))
+- Waiting for [another event](https://github.com/triggerdotdev/trigger.dev/issues/472) to fire.
+
+
+
+Tasks are so important that we've dedicated a whole section to them. See our [Tasks guide](/documentation/concepts/tasks) for more information.
+
+### Built in tasks
+
+We provide some built-in tasks that you can use in your run function:
+
+| Task | Description | Task code |
+| ------------------------------------------- | ----------------------------------------------------------------------- | ------------------------------------------------------------------------------------ |
+| [Delay](/documentation/concepts/delays) | Wait for a period of time | `await io.wait("wait", 60);` |
+| [Log](/sdk/io/logger) | Log a message | `await io.logger.log("Hello");` |
+| [Send Event](/sdk/io/sendevent) | Send an event (for eventTrigger) | `await io.sendEvent("my-event", { name: "my.event", payload: { hello: "world" } });` |
+| [Background fetch](/sdk/io/backgroundfetch) | Fetch data from a URL that can take longer that the serverless timeout. | `await io.backgroundFetch("fetch-some-data", { url: "https://example.com" });` |
+
+For a full list of built-in Tasks, see the [io SDK reference](/sdk/io). The below example makes use a few of these built-in tasks:
+
+```ts
+import { Slack } from "@trigger.dev/slack";
+
+const slack = new Slack({ id: "slack" });
+
+client.defineJob({
+ id: "my-job",
+ name: "My Job",
+ version: "1.0.0",
+ trigger: eventTrigger({
+ name: "my.event",
+ }),
+ integrations: { slack },
+ run: async (payload, io, context) => {
+ await io.logger.info("Received the my.event event", { payload });
+ await io.sendEvent("send-event", {
+ name: "other.event",
+ payload: { hello: "world" },
+ });
+
+ await io.wait("wait for 60 seconds", 60);
+
+ await io.backgroundFetch("fetch-some-data", {
+ url: "https://example.com",
+ });
+ },
+});
+```
+
+### Create your own tasks
+
+You can also create your own tasks or use tasks provided by our integration packages. See [Creating Tasks](/documentation/concepts/tasks#creating-tasks) for more information. The example below demonstrates creating tasks in 3 different ways:
+
+```ts
+import { Slack } from "@trigger.dev/slack";
+
+const slack = new Slack({ id: "slack" });
+
+client.defineJob({
+ id: "my-job",
+ name: "My Job",
+ version: "1.0.0",
+ trigger: eventTrigger({
+ name: "my.event",
+ }),
+ integrations: { slack },
+ run: async (payload, io, context) => {
+ // Use runTask with the "get-user" cacheKey, and return the user
+ const user = await io.runTask("get-user", async () => {
+ return prisma.user.findUniqueOrThrow({
+ where: {
+ id: payload.id,
+ },
+ });
+ });
+
+ // Use the Slack integration to create a task using the "post-message" cacheKey
+ const message = await io.slack.postMessage("post-message", {
+ channel: process.env.SLACK_CHANNEL_ID,
+ message: `Hello ${user.name}`,
+ });
+
+ await io.wait("wait for 10 seconds", 10);
+
+ // Use the Slack integration's runTask method to add a reaction to the message
+ await io.slack.runTask("add-reaction", async (client) => {
+ // client here is an authenticated instance of the Slack SDK
+ await client.reactions.add({
+ channel: process.env.SLACK_CHANNEL_ID,
+ name: "thumbsup",
+ timestamp: message.ts,
+ });
+ });
+ },
+});
+```
+
+## 7. Handling errors
+
+If your run function throws an error, the Job Run will fail and the error will be displayed in the Dashboard. If you'd like to retry on an error, you can do so using tasks and the `retry` option.
+
+```ts
+const user = await io.runTask(
+ "get-user",
+ async () => {
+ return prisma.user.findUniqueOrThrow({
+ where: {
+ id: payload.id,
+ },
+ });
+ },
+ {
+ retry: {
+ limit: 3,
+ factor: 2,
+ minTimeoutInMs: 1000,
+ },
+ }
+);
+```
+
+See the [retry options](/sdk/io/runtask) for more information.
+
+## 8. Skip catching internal errors
+
+We will throw some errors internally to interrupt run execution so they can be resumed later. If you put a `try/catch` block in your run code and catch these errors, your job will not work correctly. You can check if an error is an internal error using [isTriggerError()](/sdk/istriggererror):
+
+```ts
+client.defineJob({
+ run: async (payload, io, context) => {
+ try {
+ // Use runTask with the "get-user" cacheKey, and return the user
+ const user = await io.runTask("get-user", async () => {
+ return prisma.user.findUniqueOrThrow({
+ where: {
+ id: payload.id,
+ },
+ });
+ });
+ } catch (error) {
+ if (isTriggerError(error)) throw error;
+
+ // do something with your error here
+ }
+ },
+});
+```
+
+Alternatively, you can use the [io.try()](/sdk/io/try) function:
+
+```ts
+client.defineJob({
+ run: async (payload, io, context) => {
+ const result = io.try(
+ () => {
+ return io.runTask("get-user", async () => {
+ return prisma.user.findUniqueOrThrow({
+ where: {
+ id: payload.id,
+ },
+ });
+ });
+ },
+ async (error) => {
+ //you can return data from the error handler,
+ //if you wish to elegantly deal with errors
+ return {
+ success: false as const,
+ error,
+ };
+ }
+ );
+ },
+});
+```
+
+## 9. Creating tasks in a loop
+
+If you want to create tasks in a loop, you should use `for const ... of` to ensure that the tasks are created in the correct order.
+
+```ts
+client.defineJob({
+ run: async (payload, io, context) => {
+ for (const user of payload.users) {
+ await io.runTask(`update-user-${user.id}`, async () => {
+ return prisma.user.update({
+ where: {
+ id: user.id,
+ },
+ data: {
+ name: user.name,
+ },
+ });
+ });
+ }
+ },
+});
+```
+
+
+ We don't currently support running tasks in parallel so `Promise.all` will not work correctly.
+ This is something on our roadmap that we hope to support soon.
+
+
+## 10. Return data from your Job
+
+Anything you return from the `run` function will be automatically set as the run output and displayed in the Dashboard.
+
+```ts
+client.defineJob({
+ run: async (payload, io, context) => {
+ return {
+ success: true,
+ message: "Hello world",
+ };
+ },
+});
+```
+
+## Next steps
+
+We recommend exploring all of the below sections to fully understand how to create and run Jobs using Trigger.dev.
+
+
+
+ A guide for how to run your Jobs. Triggering a test Run and Triggering your Job for real
+
+
+
+
+ View example Jobs / the example jobs repo. These are a great starting
+ point for creating your own Jobs.
+
+
+ How to use the SDK. This includes all the available Tasks, triggers and
+ actions you can use.
+
+
+
+ Integrations make it easy to authenticate and use APIs.
+Learn how to use and create integrations.
+
+
diff --git a/docs/images/task.png b/docs/images/task.png
new file mode 100644
index 000000000..e45068d3f
Binary files /dev/null and b/docs/images/task.png differ
diff --git a/docs/integrations/apis/github.mdx b/docs/integrations/apis/github.mdx
index 4b00021be..a3ee2e175 100644
--- a/docs/integrations/apis/github.mdx
+++ b/docs/integrations/apis/github.mdx
@@ -5,7 +5,11 @@ sidebarTitle: Overview & authentication
## Overview
-Our GitHub integration allows you to create triggers and tasks that interact with GitHub. For examples of some of the things you can do with it, check out our Jobs Showcase:
+Our GitHub integration allows you to create triggers and tasks that interact with GitHub.
+
+Trigger jobs when events happen, such as when a new issue is added to a repo, a commit is pushed, or a pull request is opened. You can also use the integration to perform tasks such as creating issues, getting information about a repo, adding comments, and much more.
+
+For examples of some of the things you can do with it, check out our Jobs Showcase:
+ Any interaction with an external service (database or API) should be wrapped in a Task. Failing to
+ do so could result in repeated work when runs are resumed.
+
## Parameters
@@ -159,32 +162,30 @@ If the remote callback feature `options.callback` is enabled, the Promise will i
client.defineJob({
id: "alert-on-new-github-issues",
name: "Alert on new GitHub issues",
- version: "0.1.1",
+ version: "1.0.0",
trigger: github.triggers.repo({
event: events.onIssueOpened,
owner: "triggerdotdev",
repo: "trigger.dev",
}),
- integrations: {
- github,
- },
run: async (payload, io, ctx) => {
- //runTask
- const response = await io.github.runTask(
- "create-card",
- async (client) => {
- //create a project card using the underlying GitHub Integration client
- return client.rest.projects.createCard({
- column_id: 123,
- note: "test",
+ const record = await io.runTask(
+ "sync-github-issue",
+ async (task) => {
+ return prisma.githubIssues.create({
+ data: {
+ number: payload.issue.number,
+ title: payload.issue.title,
+ body: payload.issue.body,
+ url: payload.issue.html_url,
+ repo: payload.repository.full_name,
+ owner: payload.repository.owner.login,
+ },
});
},
//this is optional
- { name: "Create card", icon: "github" }
+ { name: "Sync GitHub Issue", icon: "github" }
);
-
- //log the url of the created card
- await io.logger.info(response.data.url);
},
});
```
@@ -227,14 +228,12 @@ client.defineJob({
name: "Remote Callback example",
version: "0.1.1",
trigger: eventTrigger({ name: "predict" }),
- integrations: { replicate },
run: async (payload, io, ctx) => {
- //runTask
- const prediction = await io.replicate.runTask(
+ const prediction = await io.runTask(
"create-and-await-prediction",
- async (client, task) => {
- //create a prediction using the underlying Replicate Integration client
- await client.predictions.create({
+ async (task) => {
+ //create a prediction using a Replicate SDK instance
+ await replicate.predictions.create({
...payload,
webhook: task.callbackUrl ?? "",
webhook_events_filter: ["completed"],
diff --git a/packages/core/src/schemas/api.ts b/packages/core/src/schemas/api.ts
index d95ebf65e..57dd99843 100644
--- a/packages/core/src/schemas/api.ts
+++ b/packages/core/src/schemas/api.ts
@@ -177,12 +177,14 @@ export type HttpSourceRequestHeaders = z.output;
+export const AutoYieldConfigSchema = z.object({
+ startTaskThreshold: z.number(),
+ beforeExecuteTaskThreshold: z.number(),
+ beforeCompleteTaskThreshold: z.number(),
+ afterCompleteTaskThreshold: z.number(),
+});
+
+export type AutoYieldConfig = z.infer;
+
export const RunJobBodySchema = z.object({
event: ApiEventLogSchema,
job: z.object({
@@ -485,6 +497,8 @@ export const RunJobBodySchema = z.object({
noopTasksSet: z.string().optional(),
connections: z.record(ConnectionAuthSchema).optional(),
yieldedExecutions: z.string().array().optional(),
+ runChunkExecutionLimit: z.number().optional(),
+ autoYieldConfig: AutoYieldConfigSchema.optional(),
});
export type RunJobBody = z.infer;
@@ -504,6 +518,33 @@ export const RunJobYieldExecutionErrorSchema = z.object({
export type RunJobYieldExecutionError = z.infer;
+export const RunJobAutoYieldExecutionErrorSchema = z.object({
+ status: z.literal("AUTO_YIELD_EXECUTION"),
+ location: z.string(),
+ timeRemaining: z.number(),
+ timeElapsed: z.number(),
+ limit: z.number().optional(),
+});
+
+export type RunJobAutoYieldExecutionError = z.infer;
+
+export const RunJobAutoYieldWithCompletedTaskExecutionErrorSchema = z.object({
+ status: z.literal("AUTO_YIELD_EXECUTION_WITH_COMPLETED_TASK"),
+ id: z.string(),
+ properties: z.array(DisplayPropertySchema).optional(),
+ output: z.any(),
+ data: z.object({
+ location: z.string(),
+ timeRemaining: z.number(),
+ timeElapsed: z.number(),
+ limit: z.number().optional(),
+ }),
+});
+
+export type RunJobAutoYieldWithCompletedTaskExecutionError = z.infer<
+ typeof RunJobAutoYieldWithCompletedTaskExecutionErrorSchema
+>;
+
export const RunJobInvalidPayloadErrorSchema = z.object({
status: z.literal("INVALID_PAYLOAD"),
errors: z.array(SchemaErrorSchema),
@@ -549,6 +590,8 @@ export const RunJobSuccessSchema = z.object({
export type RunJobSuccess = z.infer;
export const RunJobResponseSchema = z.discriminatedUnion("status", [
+ RunJobAutoYieldExecutionErrorSchema,
+ RunJobAutoYieldWithCompletedTaskExecutionErrorSchema,
RunJobYieldExecutionErrorSchema,
RunJobErrorSchema,
RunJobUnresolvedAuthErrorSchema,
diff --git a/packages/core/src/schemas/tasks.ts b/packages/core/src/schemas/tasks.ts
index 559dc4bab..15744bfc2 100644
--- a/packages/core/src/schemas/tasks.ts
+++ b/packages/core/src/schemas/tasks.ts
@@ -37,6 +37,7 @@ export const TaskSchema = z.object({
export const ServerTaskSchema = TaskSchema.extend({
idempotencyKey: z.string(),
attempts: z.number(),
+ forceYield: z.boolean().optional().nullable(),
});
export type ServerTask = z.infer;
diff --git a/packages/database/prisma/migrations/20231011104302_add_run_chunk_column/migration.sql b/packages/database/prisma/migrations/20231011104302_add_run_chunk_column/migration.sql
new file mode 100644
index 000000000..3e28ece9a
--- /dev/null
+++ b/packages/database/prisma/migrations/20231011104302_add_run_chunk_column/migration.sql
@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "Endpoint" ADD COLUMN "runChunkExecutionLimit" INTEGER NOT NULL DEFAULT 60;
diff --git a/packages/database/prisma/migrations/20231011134840_add_auto_yielded_executions/migration.sql b/packages/database/prisma/migrations/20231011134840_add_auto_yielded_executions/migration.sql
new file mode 100644
index 000000000..be62509c2
--- /dev/null
+++ b/packages/database/prisma/migrations/20231011134840_add_auto_yielded_executions/migration.sql
@@ -0,0 +1,14 @@
+-- CreateTable
+CREATE TABLE "JobRunAutoYieldExecution" (
+ "id" TEXT NOT NULL,
+ "runId" TEXT NOT NULL,
+ "timeRemaining" INTEGER NOT NULL,
+ "timeElapsed" INTEGER NOT NULL,
+ "limit" INTEGER NOT NULL,
+ "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+ CONSTRAINT "JobRunAutoYieldExecution_pkey" PRIMARY KEY ("id")
+);
+
+-- AddForeignKey
+ALTER TABLE "JobRunAutoYieldExecution" ADD CONSTRAINT "JobRunAutoYieldExecution_runId_fkey" FOREIGN KEY ("runId") REFERENCES "JobRun"("id") ON DELETE CASCADE ON UPDATE CASCADE;
diff --git a/packages/database/prisma/migrations/20231011141302_add_location_to_auto_yield_executions/migration.sql b/packages/database/prisma/migrations/20231011141302_add_location_to_auto_yield_executions/migration.sql
new file mode 100644
index 000000000..f8f1775b4
--- /dev/null
+++ b/packages/database/prisma/migrations/20231011141302_add_location_to_auto_yield_executions/migration.sql
@@ -0,0 +1,8 @@
+/*
+ Warnings:
+
+ - Added the required column `location` to the `JobRunAutoYieldExecution` table without a default value. This is not possible if the table is not empty.
+
+*/
+-- AlterTable
+ALTER TABLE "JobRunAutoYieldExecution" ADD COLUMN "location" TEXT NOT NULL;
diff --git a/packages/database/prisma/migrations/20231011145406_change_run_chunk_execution_limit_default/migration.sql b/packages/database/prisma/migrations/20231011145406_change_run_chunk_execution_limit_default/migration.sql
new file mode 100644
index 000000000..bd19a6bb1
--- /dev/null
+++ b/packages/database/prisma/migrations/20231011145406_change_run_chunk_execution_limit_default/migration.sql
@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "Endpoint" ALTER COLUMN "runChunkExecutionLimit" SET DEFAULT 60000;
diff --git a/packages/database/prisma/migrations/20231011213532_add_auto_yield_threshold_settings_to_endpoints/migration.sql b/packages/database/prisma/migrations/20231011213532_add_auto_yield_threshold_settings_to_endpoints/migration.sql
new file mode 100644
index 000000000..260f0fcdf
--- /dev/null
+++ b/packages/database/prisma/migrations/20231011213532_add_auto_yield_threshold_settings_to_endpoints/migration.sql
@@ -0,0 +1,5 @@
+-- AlterTable
+ALTER TABLE "Endpoint" ADD COLUMN "afterCompleteTaskThreshold" INTEGER NOT NULL DEFAULT 750,
+ADD COLUMN "beforeCompleteTaskThreshold" INTEGER NOT NULL DEFAULT 750,
+ADD COLUMN "beforeExecuteTaskThreshold" INTEGER NOT NULL DEFAULT 1500,
+ADD COLUMN "startTaskThreshold" INTEGER NOT NULL DEFAULT 750;
diff --git a/packages/database/prisma/migrations/20231019123406_add_force_yield_immediately_to_runs/migration.sql b/packages/database/prisma/migrations/20231019123406_add_force_yield_immediately_to_runs/migration.sql
new file mode 100644
index 000000000..4ede43f10
--- /dev/null
+++ b/packages/database/prisma/migrations/20231019123406_add_force_yield_immediately_to_runs/migration.sql
@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "JobRun" ADD COLUMN "forceYieldImmediately" BOOLEAN NOT NULL DEFAULT false;
diff --git a/packages/database/prisma/migrations/20231020145127_add_sdk_version_to_endpoints/migration.sql b/packages/database/prisma/migrations/20231020145127_add_sdk_version_to_endpoints/migration.sql
new file mode 100644
index 000000000..3180c6531
--- /dev/null
+++ b/packages/database/prisma/migrations/20231020145127_add_sdk_version_to_endpoints/migration.sql
@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "Endpoint" ADD COLUMN "sdkVersion" TEXT NOT NULL DEFAULT 'unknown';
diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma
index 7f1d21918..302be3bf8 100644
--- a/packages/database/prisma/schema.prisma
+++ b/packages/database/prisma/schema.prisma
@@ -368,6 +368,13 @@ model Endpoint {
indexingHookIdentifier String?
version String @default("unknown")
+ sdkVersion String @default("unknown")
+
+ runChunkExecutionLimit Int @default(60000)
+ startTaskThreshold Int @default(750)
+ beforeExecuteTaskThreshold Int @default(1500)
+ beforeCompleteTaskThreshold Int @default(750)
+ afterCompleteTaskThreshold Int @default(750)
jobVersions JobVersion[]
jobRuns JobRun[]
@@ -726,11 +733,14 @@ model JobRun {
yieldedExecutions String[]
+ forceYieldImmediately Boolean @default(false)
+
tasks Task[]
runConnections RunConnection[]
missingConnections MissingConnection[]
executions JobRunExecution[]
statuses JobRunStatusRecord[]
+ autoYieldExecution JobRunAutoYieldExecution[]
}
enum JobRunStatus {
@@ -748,6 +758,20 @@ enum JobRunStatus {
INVALID_PAYLOAD
}
+model JobRunAutoYieldExecution {
+ id String @id @default(cuid())
+
+ run JobRun @relation(fields: [runId], references: [id], onDelete: Cascade, onUpdate: Cascade)
+ runId String
+
+ timeRemaining Int
+ timeElapsed Int
+ limit Int
+ location String
+
+ createdAt DateTime @default(now())
+}
+
model JobRunExecution {
id String @id @default(cuid())
diff --git a/packages/trigger-sdk/src/errors.ts b/packages/trigger-sdk/src/errors.ts
index 9e2aeec0e..d8fcf404d 100644
--- a/packages/trigger-sdk/src/errors.ts
+++ b/packages/trigger-sdk/src/errors.ts
@@ -1,3 +1,4 @@
+import { DisplayProperty } from "@trigger.dev/core";
import { ErrorWithStack, SchemaError, ServerTask } from "@trigger.dev/core";
export class ResumeWithTaskError {
@@ -20,6 +21,23 @@ export class YieldExecutionError {
constructor(public key: string) {}
}
+export class AutoYieldExecutionError {
+ constructor(
+ public location: string,
+ public timeRemaining: number,
+ public timeElapsed: number
+ ) {}
+}
+
+export class AutoYieldWithCompletedTaskExecutionError {
+ constructor(
+ public id: string,
+ public properties: DisplayProperty[] | undefined,
+ public output: any,
+ public data: { location: string; timeRemaining: number; timeElapsed: number }
+ ) {}
+}
+
export class ParsedPayloadSchemaError {
constructor(public schemaErrors: SchemaError[]) {}
}
@@ -32,11 +50,19 @@ export class ParsedPayloadSchemaError {
*/
export function isTriggerError(
err: unknown
-): err is ResumeWithTaskError | RetryWithTaskError | CanceledWithTaskError {
+): err is
+ | ResumeWithTaskError
+ | RetryWithTaskError
+ | CanceledWithTaskError
+ | YieldExecutionError
+ | AutoYieldExecutionError
+ | AutoYieldWithCompletedTaskExecutionError {
return (
err instanceof ResumeWithTaskError ||
err instanceof RetryWithTaskError ||
err instanceof CanceledWithTaskError ||
- err instanceof YieldExecutionError
+ err instanceof YieldExecutionError ||
+ err instanceof AutoYieldExecutionError ||
+ err instanceof AutoYieldWithCompletedTaskExecutionError
);
}
diff --git a/packages/trigger-sdk/src/io.ts b/packages/trigger-sdk/src/io.ts
index b4dacfc5e..de9a77531 100644
--- a/packages/trigger-sdk/src/io.ts
+++ b/packages/trigger-sdk/src/io.ts
@@ -23,6 +23,8 @@ import { AsyncLocalStorage } from "node:async_hooks";
import { webcrypto } from "node:crypto";
import { ApiClient } from "./apiClient";
import {
+ AutoYieldExecutionError,
+ AutoYieldWithCompletedTaskExecutionError,
CanceledWithTaskError,
ResumeWithTaskError,
RetryWithTaskError,
@@ -45,6 +47,7 @@ export type IOOptions = {
apiClient: ApiClient;
client: TriggerClient;
context: TriggerContext;
+ timeOrigin: number;
logger?: Logger;
logLevel?: LogLevel;
jobLogger?: Logger;
@@ -54,6 +57,7 @@ export type IOOptions = {
yieldedExecutions?: Array;
noopTasksSet?: string;
serverVersion?: string | null;
+ executionTimeout?: number;
};
type JsonPrimitive = string | number | boolean | null | undefined | Date | symbol;
@@ -96,6 +100,8 @@ export class IO {
private _noopTasksBloomFilter: BloomFilter | undefined;
private _stats: IOStats;
private _serverVersion: string;
+ private _timeOrigin: number;
+ private _executionTimeout?: number;
get stats() {
return this._stats;
@@ -109,6 +115,8 @@ export class IO {
this._cachedTasks = new Map();
this._jobLogger = options.jobLogger;
this._jobLogLevel = options.jobLogLevel;
+ this._timeOrigin = options.timeOrigin;
+ this._executionTimeout = options.executionTimeout;
this._stats = {
initialCachedTasks: 0,
@@ -205,11 +213,11 @@ export class IO {
}
/** `io.wait()` waits for the specified amount of time before continuing the Job. Delays work even if you're on a serverless platform with timeouts, or if your server goes down. They utilize [resumability](https://trigger.dev/docs/documentation/concepts/resumability) to ensure that the Run can be resumed after the delay.
- * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+ * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
* @param seconds The number of seconds to wait. This can be very long, serverless timeouts are not an issue.
*/
- async wait(key: string | any[], seconds: number) {
- return await this.runTask(key, async (task) => {}, {
+ async wait(cacheKey: string | any[], seconds: number) {
+ return await this.runTask(cacheKey, async (task) => {}, {
name: "wait",
icon: "clock",
params: { seconds },
@@ -220,7 +228,7 @@ export class IO {
}
/** `io.createStatus()` allows you to set a status with associated data during the Run. Statuses can be used by your UI using the react package
- * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+ * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
* @param initialStatus The initial status you want this status to have. You can update it during the rub using the returned object.
* @returns a TriggerStatus object that you can call `update()` on, to update the status.
* @example
@@ -252,17 +260,17 @@ export class IO {
* ```
*/
async createStatus(
- key: IntegrationTaskKey,
+ cacheKey: IntegrationTaskKey,
initialStatus: InitialStatusUpdate
): Promise {
- const id = typeof key === "string" ? key : key.join("-");
+ const id = typeof cacheKey === "string" ? cacheKey : cacheKey.join("-");
const status = new TriggerStatus(id, this);
- await status.update(key, initialStatus);
+ await status.update(cacheKey, initialStatus);
return status;
}
/** `io.backgroundFetch()` fetches data from a URL that can take longer that the serverless timeout. The actual `fetch` request is performed on the Trigger.dev platform, and the response is sent back to you.
- * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+ * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
* @param url The URL to fetch from.
* @param requestInit The options for the request
* @param retry The options for retrying the request if it fails
@@ -273,7 +281,7 @@ export class IO {
* - Wildcards: 2xx, 3xx, 4xx, 5xx
*/
async backgroundFetch(
- key: string | any[],
+ cacheKey: string | any[],
url: string,
requestInit?: FetchRequestInit,
retry?: FetchRetryOptions
@@ -281,7 +289,7 @@ export class IO {
const urlObject = new URL(url);
return (await this.runTask(
- key,
+ cacheKey,
async (task) => {
return task.output;
},
@@ -311,13 +319,13 @@ export class IO {
}
/** `io.sendEvent()` allows you to send an event from inside a Job run. The sent even will trigger any Jobs that are listening for that event (based on the name).
- * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+ * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
* @param event The event to send. The event name must match the name of the event that your Jobs are listening for.
* @param options Options for sending the event.
*/
- async sendEvent(key: string | any[], event: SendEvent, options?: SendEventOptions) {
+ async sendEvent(cacheKey: string | any[], event: SendEvent, options?: SendEventOptions) {
return await this.runTask(
- key,
+ cacheKey,
async (task) => {
return await this._triggerClient.sendEvent(event, options);
},
@@ -335,9 +343,9 @@ export class IO {
);
}
- async getEvent(key: string | any[], id: string) {
+ async getEvent(cacheKey: string | any[], id: string) {
return await this.runTask(
- key,
+ cacheKey,
async (task) => {
return await this._triggerClient.getEvent(id);
},
@@ -355,13 +363,13 @@ export class IO {
}
/** `io.cancelEvent()` allows you to cancel an event that was previously sent with `io.sendEvent()`. This will prevent any Jobs from running that are listening for that event if the event was sent with a delay
- * @param key
+ * @param cacheKey
* @param eventId
* @returns
*/
- async cancelEvent(key: string | any[], eventId: string) {
+ async cancelEvent(cacheKey: string | any[], eventId: string) {
return await this.runTask(
- key,
+ cacheKey,
async (task) => {
return await this._triggerClient.cancelEvent(eventId);
},
@@ -380,9 +388,12 @@ export class IO {
);
}
- async updateSource(key: string | any[], options: { key: string } & UpdateTriggerSourceBodyV2) {
+ async updateSource(
+ cacheKey: string | any[],
+ options: { key: string } & UpdateTriggerSourceBodyV2
+ ) {
return this.runTask(
- key,
+ cacheKey,
async (task) => {
return await this._apiClient.updateSource(this._triggerClient.id, options.key, options);
},
@@ -404,7 +415,7 @@ export class IO {
}
/** `io.registerInterval()` allows you to register a [DynamicSchedule](https://trigger.dev/docs/sdk/dynamicschedule) that will trigger any jobs it's attached to on a regular interval.
- * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+ * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
* @param dynamicSchedule The [DynamicSchedule](https://trigger.dev/docs/sdk/dynamicschedule) to register a new schedule on.
* @param id A unique id for the interval. This is used to identify and unregister the interval later.
* @param options The options for the interval.
@@ -412,13 +423,13 @@ export class IO {
* @deprecated Use `DynamicSchedule.register` instead.
*/
async registerInterval(
- key: string | any[],
+ cacheKey: string | any[],
dynamicSchedule: DynamicSchedule,
id: string,
options: IntervalOptions
) {
return await this.runTask(
- key,
+ cacheKey,
async (task) => {
return dynamicSchedule.register(id, {
type: "interval",
@@ -438,14 +449,14 @@ export class IO {
}
/** `io.unregisterInterval()` allows you to unregister a [DynamicSchedule](https://trigger.dev/docs/sdk/dynamicschedule) that was previously registered with `io.registerInterval()`.
- * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+ * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
* @param dynamicSchedule The [DynamicSchedule](https://trigger.dev/docs/sdk/dynamicschedule) to unregister a schedule on.
* @param id A unique id for the interval. This is used to identify and unregister the interval later.
* @deprecated Use `DynamicSchedule.unregister` instead.
*/
- async unregisterInterval(key: string | any[], dynamicSchedule: DynamicSchedule, id: string) {
+ async unregisterInterval(cacheKey: string | any[], dynamicSchedule: DynamicSchedule, id: string) {
return await this.runTask(
- key,
+ cacheKey,
async (task) => {
return dynamicSchedule.unregister(id);
},
@@ -460,20 +471,20 @@ export class IO {
}
/** `io.registerCron()` allows you to register a [DynamicSchedule](https://trigger.dev/docs/sdk/dynamicschedule) that will trigger any jobs it's attached to on a regular CRON schedule.
- * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+ * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
* @param dynamicSchedule The [DynamicSchedule](https://trigger.dev/docs/sdk/dynamicschedule) to register a new schedule on.
* @param id A unique id for the schedule. This is used to identify and unregister the schedule later.
* @param options The options for the CRON schedule.
* @deprecated Use `DynamicSchedule.register` instead.
*/
async registerCron(
- key: string | any[],
+ cacheKey: string | any[],
dynamicSchedule: DynamicSchedule,
id: string,
options: CronOptions
) {
return await this.runTask(
- key,
+ cacheKey,
async (task) => {
return dynamicSchedule.register(id, {
type: "cron",
@@ -493,14 +504,14 @@ export class IO {
}
/** `io.unregisterCron()` allows you to unregister a [DynamicSchedule](https://trigger.dev/docs/sdk/dynamicschedule) that was previously registered with `io.registerCron()`.
- * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+ * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
* @param dynamicSchedule The [DynamicSchedule](https://trigger.dev/docs/sdk/dynamicschedule) to unregister a schedule on.
* @param id A unique id for the interval. This is used to identify and unregister the interval later.
* @deprecated Use `DynamicSchedule.unregister` instead.
*/
- async unregisterCron(key: string | any[], dynamicSchedule: DynamicSchedule, id: string) {
+ async unregisterCron(cacheKey: string | any[], dynamicSchedule: DynamicSchedule, id: string) {
return await this.runTask(
- key,
+ cacheKey,
async (task) => {
return dynamicSchedule.unregister(id);
},
@@ -515,7 +526,7 @@ export class IO {
}
/** `io.registerTrigger()` allows you to register a [DynamicTrigger](https://trigger.dev/docs/sdk/dynamictrigger) with the specified trigger params.
- * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+ * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
* @param trigger The [DynamicTrigger](https://trigger.dev/docs/sdk/dynamictrigger) to register.
* @param id A unique id for the trigger. This is used to identify and unregister the trigger later.
* @param params The params for the trigger.
@@ -524,13 +535,13 @@ export class IO {
async registerTrigger<
TTrigger extends DynamicTrigger, ExternalSource>,
>(
- key: string | any[],
+ cacheKey: string | any[],
trigger: TTrigger,
id: string,
params: ExternalSourceParams
): Promise<{ id: string; key: string } | undefined> {
return await this.runTask(
- key,
+ cacheKey,
async (task) => {
const registration = await this.runTask(
"register-source",
@@ -558,13 +569,13 @@ export class IO {
);
}
- async getAuth(key: string | any[], clientId?: string): Promise {
+ async getAuth(cacheKey: string | any[], clientId?: string): Promise {
if (!clientId) {
return;
}
return this.runTask(
- key,
+ cacheKey,
async (task) => {
return await this._triggerClient.getAuth(clientId);
},
@@ -574,29 +585,33 @@ export class IO {
/** `io.runTask()` allows you to run a [Task](https://trigger.dev/docs/documentation/concepts/tasks) from inside a Job run. A Task is a resumable unit of a Run that can be retried, resumed and is logged. [Integrations](https://trigger.dev/docs/integrations) use Tasks internally to perform their actions.
*
- * @param key Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
+ * @param cacheKey Should be a stable and unique key inside the `run()`. See [resumability](https://trigger.dev/docs/documentation/concepts/resumability) for more information.
* @param callback The callback that will be called when the Task is run. The callback receives the Task and the IO as parameters.
* @param options The options of how you'd like to run and log the Task.
* @param onError The callback that will be called when the Task fails. The callback receives the error, the Task and the IO as parameters. If you wish to retry then return an object with a `retryAt` property.
* @returns A Promise that resolves with the returned value of the callback.
*/
async runTask | void>(
- key: string | any[],
+ cacheKey: string | any[],
callback: (task: ServerTask, io: IO) => Promise,
options?: RunTaskOptions,
onError?: RunTaskErrorCallback
): Promise {
+ this.#detectAutoYield("start_task", 500);
+
const parentId = this._taskStorage.getStore()?.taskId;
if (parentId) {
this._logger.debug("Using parent task", {
parentId,
- key,
+ cacheKey,
options,
});
}
- const idempotencyKey = await generateIdempotencyKey([this._id, parentId ?? "", key].flat());
+ const idempotencyKey = await generateIdempotencyKey(
+ [this._id, parentId ?? "", cacheKey].flat()
+ );
const cachedTask = this._cachedTasks.get(idempotencyKey);
@@ -626,7 +641,7 @@ export class IO {
this._id,
{
idempotencyKey,
- displayKey: typeof key === "string" ? key : undefined,
+ displayKey: typeof cacheKey === "string" ? cacheKey : undefined,
noop: false,
...(options ?? {}),
parentId,
@@ -641,6 +656,14 @@ export class IO {
? response.body.task
: response.body;
+ if (task.forceYield) {
+ this._logger.debug("Forcing yield after run task", {
+ idempotencyKey,
+ });
+
+ this.#forceYield("after_run_task");
+ }
+
if (response.version === API_VERSIONS.LAZY_LOADED_CACHED_TASKS) {
this._cachedTasksCursor = response.body.cachedTasks?.cursor;
@@ -694,6 +717,8 @@ export class IO {
throw new Error(task.error ?? task?.output ? JSON.stringify(task.output) : "Task errored");
}
+ this.#detectAutoYield("before_execute_task", 1500);
+
const executeTask = async () => {
try {
const result = await callback(task, this);
@@ -713,17 +738,29 @@ export class IO {
task,
});
+ this.#detectAutoYield("before_complete_task", 500, task, output);
+
const completedTask = await this._apiClient.completeTask(this._id, task.id, {
output: output ?? undefined,
properties: task.outputProperties ?? undefined,
});
+ if (completedTask.forceYield) {
+ this._logger.debug("Forcing yield after task completed", {
+ idempotencyKey,
+ });
+
+ this.#forceYield("after_complete_task");
+ }
+
this._stats.executedTasks++;
if (completedTask.status === "CANCELED") {
throw new CanceledWithTaskError(completedTask);
}
+ this.#detectAutoYield("after_complete_task", 500);
+
return output;
} catch (error) {
if (isTriggerError(error)) {
@@ -818,7 +855,7 @@ export class IO {
/**
* `io.yield()` allows you to yield execution of the current run and resume it in a new function execution. Similar to `io.wait()` but does not create a task and resumes execution immediately.
*/
- yield(key: string) {
+ yield(cacheKey: string) {
if (!supportsFeature("yieldExecution", this._serverVersion)) {
console.warn(
"[trigger.dev] io.yield() is not support by the version of the Trigger.dev server you are using, you will need to upgrade your self-hosted Trigger.dev instance."
@@ -827,11 +864,11 @@ export class IO {
return;
}
- if (this._yieldedExecutions.includes(key)) {
+ if (this._yieldedExecutions.includes(cacheKey)) {
return;
}
- throw new YieldExecutionError(key);
+ throw new YieldExecutionError(cacheKey);
}
/**
@@ -863,6 +900,47 @@ export class IO {
#addToCachedTasks(task: ServerTask) {
this._cachedTasks.set(task.idempotencyKey, task);
}
+
+ #detectAutoYield(location: string, threshold: number = 1500, task?: ServerTask, output?: any) {
+ const timeRemaining = this.#getRemainingTimeInMillis();
+
+ if (timeRemaining && timeRemaining < threshold) {
+ if (task) {
+ throw new AutoYieldWithCompletedTaskExecutionError(
+ task.id,
+ task.outputProperties ?? [],
+ output,
+ {
+ location,
+ timeRemaining,
+ timeElapsed: this.#getTimeElapsed(),
+ }
+ );
+ } else {
+ throw new AutoYieldExecutionError(location, timeRemaining, this.#getTimeElapsed());
+ }
+ }
+ }
+
+ #forceYield(location: string) {
+ const timeRemaining = this.#getRemainingTimeInMillis();
+
+ if (timeRemaining) {
+ throw new AutoYieldExecutionError(location, timeRemaining, this.#getTimeElapsed());
+ }
+ }
+
+ #getTimeElapsed() {
+ return performance.now() - this._timeOrigin;
+ }
+
+ #getRemainingTimeInMillis() {
+ if (this._executionTimeout) {
+ return this._executionTimeout - (performance.now() - this._timeOrigin);
+ }
+
+ return undefined;
+ }
}
// Generate a stable idempotency key for the key material, using a stable json stringification
diff --git a/packages/trigger-sdk/src/triggerClient.ts b/packages/trigger-sdk/src/triggerClient.ts
index b1766f164..bbb5e8b21 100644
--- a/packages/trigger-sdk/src/triggerClient.ts
+++ b/packages/trigger-sdk/src/triggerClient.ts
@@ -33,6 +33,8 @@ import {
} from "@trigger.dev/core";
import { ApiClient } from "./apiClient";
import {
+ AutoYieldExecutionError,
+ AutoYieldWithCompletedTaskExecutionError,
CanceledWithTaskError,
ParsedPayloadSchemaError,
ResumeWithTaskError,
@@ -63,6 +65,8 @@ const registerSourceEvent: EventSpecification = {
parsePayload: RegisterSourceEventSchemaV2.parse,
};
+import * as packageJson from "../package.json";
+
export type TriggerClientOptions = {
/** The `id` property is used to uniquely identify the client.
*/
@@ -132,7 +136,10 @@ export class TriggerClient {
]);
}
- async handleRequest(request: Request): Promise {
+ async handleRequest(
+ request: Request,
+ timeOrigin: number = performance.now()
+ ): Promise {
this.#internalLogger.debug("handling request", {
url: request.url,
headers: Object.fromEntries(request.headers.entries()),
@@ -154,7 +161,7 @@ export class TriggerClient {
body: {
message: "Unauthorized: client missing apiKey",
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
case "missing-header": {
@@ -163,7 +170,7 @@ export class TriggerClient {
body: {
message: "Unauthorized: missing x-trigger-api-key header",
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
case "unauthorized": {
@@ -172,7 +179,7 @@ export class TriggerClient {
body: {
message: `Forbidden: client apiKey mismatch: Make sure you are using the correct API Key for your environment`,
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
}
@@ -183,7 +190,7 @@ export class TriggerClient {
body: {
message: "Method not allowed (only POST is allowed)",
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
@@ -195,7 +202,7 @@ export class TriggerClient {
body: {
message: "Missing x-trigger-action header",
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
@@ -210,7 +217,7 @@ export class TriggerClient {
ok: false,
error: "Missing endpoint ID",
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
@@ -221,7 +228,7 @@ export class TriggerClient {
ok: false,
error: `Endpoint ID mismatch error. Expected ${this.id}, got ${endpointId}`,
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
@@ -230,7 +237,7 @@ export class TriggerClient {
body: {
ok: true,
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
case "INDEX_ENDPOINT": {
@@ -255,7 +262,7 @@ export class TriggerClient {
return {
status: 200,
body,
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
case "INITIALIZE_TRIGGER": {
@@ -285,7 +292,7 @@ export class TriggerClient {
return {
status: 200,
body: dynamicTrigger.registeredTriggerForParams(body.data.params),
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
case "EXECUTE_JOB": {
@@ -312,12 +319,12 @@ export class TriggerClient {
};
}
- const results = await this.#executeJob(execution.data, job, triggerVersion);
+ const results = await this.#executeJob(execution.data, job, timeOrigin, triggerVersion);
return {
status: 200,
body: results,
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
case "PREPROCESS_RUN": {
@@ -352,7 +359,7 @@ export class TriggerClient {
abort: results.abort,
properties: results.properties,
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
case "DELIVER_HTTP_SOURCE_REQUEST": {
@@ -418,7 +425,7 @@ export class TriggerClient {
response,
metadata,
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
case "VALIDATE": {
@@ -428,7 +435,22 @@ export class TriggerClient {
ok: true,
endpointId: this.id,
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
+ };
+ }
+ case "PROBE_EXECUTION_TIMEOUT": {
+ const json = await request.json();
+ // Keep this request open for max 15 minutes so the server can detect when the function execution limit is exceeded
+ const timeout = json?.timeout ?? 15 * 60 * 1000;
+
+ await new Promise((resolve) => setTimeout(resolve, timeout));
+
+ return {
+ status: 200,
+ body: {
+ ok: true,
+ },
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
}
@@ -438,7 +460,7 @@ export class TriggerClient {
body: {
message: "Method not allowed",
},
- headers: this.#standardResponseHeaders,
+ headers: this.#standardResponseHeaders(timeOrigin),
};
}
@@ -690,6 +712,7 @@ export class TriggerClient {
async #executeJob(
body: RunJobBody,
job: Job, Record>,
+ timeOrigin: number,
triggerVersion: string | null
): Promise {
this.#internalLogger.debug("executing job", {
@@ -716,6 +739,8 @@ export class TriggerClient {
? new Logger(job.id, job.logLevel ?? this.#options.logLevel ?? "info")
: undefined,
serverVersion: triggerVersion,
+ timeOrigin,
+ executionTimeout: body.runChunkExecutionLimit,
});
const resolvedConnections = await this.#resolveConnections(
@@ -756,6 +781,29 @@ export class TriggerClient {
this.#logIOStats(io.stats);
}
+ if (error instanceof AutoYieldExecutionError) {
+ return {
+ status: "AUTO_YIELD_EXECUTION",
+ location: error.location,
+ timeRemaining: error.timeRemaining,
+ timeElapsed: error.timeElapsed,
+ limit: body.runChunkExecutionLimit,
+ };
+ }
+
+ if (error instanceof AutoYieldWithCompletedTaskExecutionError) {
+ return {
+ status: "AUTO_YIELD_EXECUTION_WITH_COMPLETED_TASK",
+ id: error.id,
+ properties: error.properties,
+ output: error.output,
+ data: {
+ ...error.data,
+ limit: body.runChunkExecutionLimit,
+ },
+ };
+ }
+
if (error instanceof YieldExecutionError) {
return { status: "YIELD_EXECUTION", key: error.key };
}
@@ -1158,9 +1206,11 @@ export class TriggerClient {
});
}
- get #standardResponseHeaders() {
+ #standardResponseHeaders(start: number) {
return {
"Trigger-Version": API_VERSIONS.LAZY_LOADED_CACHED_TASKS,
+ "Trigger-SDK-Version": packageJson.version,
+ "X-Trigger-Request-Timing": `dur=${performance.now() - start / 1000.0}`,
};
}
}
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 870043aab..6d4c913c4 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -65,7 +65,6 @@ importers:
'@codemirror/view': ^6.5.0
'@conform-to/react': ^0.6.1
'@conform-to/zod': ^0.6.1
- '@godaddy/terminus': ^4.12.1
'@headlessui/react': ^1.7.8
'@heroicons/react': ^2.0.12
'@highlight-run/node': ^3.1.0
@@ -211,7 +210,6 @@ importers:
'@codemirror/view': 6.7.2
'@conform-to/react': 0.6.1_react@18.2.0
'@conform-to/zod': 0.6.1_zod@3.22.3
- '@godaddy/terminus': 4.12.1
'@headlessui/react': 1.7.8_biqbaboplfbrettd7655fr4n2y
'@heroicons/react': 2.0.13_react@18.2.0
'@highlight-run/node': 3.1.0
@@ -6906,12 +6904,6 @@ packages:
resolution: {integrity: sha512-k2Ty1JcVojjJFwrg/ThKi2ujJ7XNLYaFGNB/bWT9wGR+oSMJHMa5w+CUq6p/pVrKeNNgA7pCqEcjSnHVoqJQFw==}
dev: true
- /@godaddy/terminus/4.12.1:
- resolution: {integrity: sha512-Tm+wVu1/V37uZXcT7xOhzdpFoovQReErff8x3y82k6YyWa1gzxWBjTyrx4G2enjEqoXPnUUmJ3MOmwH+TiP6Sw==}
- dependencies:
- stoppable: 1.1.0
- dev: false
-
/@graphile/logger/0.2.0:
resolution: {integrity: sha512-jjcWBokl9eb1gVJ85QmoaQ73CQ52xAaOCF29ukRbYNl6lY+ts0ErTaDYOBlejcbUs2OpaiqYLO5uDhyLFzWw4w==}
dev: false
@@ -18795,7 +18787,7 @@ packages:
eslint-import-resolver-webpack:
optional: true
dependencies:
- '@typescript-eslint/parser': 5.59.6_eslint@8.42.0
+ '@typescript-eslint/parser': 5.59.6_binxsscxvozjxebftqdoazsxm4
debug: 3.2.7
eslint: 8.42.0
eslint-import-resolver-node: 0.3.7
@@ -18880,7 +18872,7 @@ packages:
'@typescript-eslint/parser':
optional: true
dependencies:
- '@typescript-eslint/parser': 5.59.6_eslint@8.42.0
+ '@typescript-eslint/parser': 5.59.6_binxsscxvozjxebftqdoazsxm4
array-includes: 3.1.6
array.prototype.flat: 1.3.1
array.prototype.flatmap: 1.3.1
@@ -28918,11 +28910,6 @@ packages:
dependencies:
bl: 5.1.0
- /stoppable/1.1.0:
- resolution: {integrity: sha512-KXDYZ9dszj6bzvnEMRYvxgeTHU74QBFL54XKtP3nyMuJ81CFYtABZ3bAzL2EdFUaEwJOBOgENyFj3R7oTzDyyw==}
- engines: {node: '>=4', npm: '>=6'}
- dev: false
-
/store2/2.14.2:
resolution: {integrity: sha512-siT1RiqlfQnGqgT/YzXVUNsom9S0H1OX+dpdGN1xkyYATo4I6sep5NmsRD/40s3IIOvlCq6akxkqG82urIZW1w==}
dev: true
diff --git a/references/job-catalog/package.json b/references/job-catalog/package.json
index 0fcca5422..0059ee0e2 100644
--- a/references/job-catalog/package.json
+++ b/references/job-catalog/package.json
@@ -27,6 +27,7 @@
"redacted": "nodemon --watch src/redacted.ts -r tsconfig-paths/register -r dotenv/config src/redacted.ts",
"replicate": "nodemon --watch src/replicate.ts -r tsconfig-paths/register -r dotenv/config src/replicate.ts",
"misconfigured": "nodemon --watch src/misconfigured.ts -r tsconfig-paths/register -r dotenv/config src/misconfigured.ts",
+ "auto-yield": "nodemon --watch src/auto-yield.ts -r tsconfig-paths/register -r dotenv/config src/auto-yield.ts",
"dev:trigger": "trigger-cli dev --port 8080"
},
"dependencies": {
@@ -61,4 +62,4 @@
"ts-node": "^10.9.1",
"tsconfig-paths": "^3.14.1"
}
-}
+}
\ No newline at end of file
diff --git a/references/job-catalog/src/auto-yield.ts b/references/job-catalog/src/auto-yield.ts
new file mode 100644
index 000000000..b097979af
--- /dev/null
+++ b/references/job-catalog/src/auto-yield.ts
@@ -0,0 +1,83 @@
+import { createExpressServer } from "@trigger.dev/express";
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+
+export const client = new TriggerClient({
+ id: "job-catalog",
+ apiKey: process.env["TRIGGER_API_KEY"],
+ apiUrl: process.env["TRIGGER_API_URL"],
+ verbose: true,
+ ioLogLocalEnabled: true,
+});
+
+client.defineJob({
+ id: "auto-yield-1",
+ name: "Auto Yield 1",
+ version: "1.0.0",
+ trigger: eventTrigger({
+ name: "auto.yield.1",
+ }),
+ run: async (payload, io, ctx) => {
+ await io.runTask("initial-long-task", async (task) => {
+ await new Promise((resolve) => setTimeout(resolve, 51000)); // 51 seconds
+
+ return {
+ message: "initial-long-task",
+ };
+ });
+
+ for (let i = 0; i < payload.iterations; i++) {
+ await io.runTask(`task.${i}`, async (task) => {
+ // Create a random number between 250 and 1250
+ const random = Math.floor(Math.random() * 1000) + 250;
+
+ await new Promise((resolve) => setTimeout(resolve, random));
+
+ await fetch(payload.url, {
+ method: "POST",
+ headers: {
+ "Content-Type": "application/json",
+ },
+ body: JSON.stringify({
+ message: `task.${i}`,
+ random,
+ idempotencyKey: task.idempotencyKey,
+ runId: ctx.run.id,
+ }),
+ });
+
+ return {
+ message: `task.${i}`,
+ random,
+ };
+ });
+ }
+ },
+});
+
+client.defineJob({
+ id: "auto-yield-2",
+ name: "Auto Yield 2",
+ version: "1.0.0",
+ trigger: eventTrigger({
+ name: "auto.yield.2",
+ }),
+ run: async (payload, io, ctx) => {
+ await io.runTask("long-task-1", async (task) => {
+ await new Promise((resolve) => setTimeout(resolve, 10000));
+
+ return {
+ message: "long-task-1",
+ };
+ });
+
+ await io.runTask("long-task-2", async (task) => {
+ await new Promise((resolve) => setTimeout(resolve, 10000));
+
+ return {
+ message: "long-task-2",
+ };
+ });
+ },
+});
+
+createExpressServer(client);
-