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

v3: Fixes for using (batch)triggerAndWait with idempotency keys (#1043)

Browse Source 
* Fixes various issues with triggerAndWait and batchTriggerAndWait

When using idempotency keys, triggerAndWait and batchTriggerAndWait will still work even if the existing runs have already been completed (or even partially completed, in the case of batchTriggerAndWait)

- TaskRunExecutionResult.id is now the run friendlyId, not the attempt friendlyId
- A single TaskRun can now have many batchItems, in the case of batchTriggerAndWait while using idempotency keys
- A run’s idempotencyKey is now added to the ctx as well as the TaskEvent and displayed in the span view
- When resolving batchTriggerAndWait, the runtimes no longer reject promises, leading to an error in the parent task

* Remove the default queue concurrency limit as we now have env and org concurrency limits

* Use the run friendlyId in the completion result id

* Added some error logging
This commit is contained in:
Eric Allam
2024-04-19 10:54:43 +01:00
committed by GitHub
parent e7bd1ee676
commit 26093896d2
47 changed files with 720 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/sharp-emus-compare.md
+13
View File
@@ -0,0 +1,13 @@
---
"@trigger.dev/sdk": patch
"trigger.dev": patch
"@trigger.dev/core": patch
---
When using idempotency keys, triggerAndWait and batchTriggerAndWait will still work even if the existing runs have already been completed (or even partially completed, in the case of batchTriggerAndWait)
- TaskRunExecutionResult.id is now the run friendlyId, not the attempt friendlyId
- A single TaskRun can now have many batchItems, in the case of batchTriggerAndWait while using idempotency keys
- A run’s idempotencyKey is now added to the ctx as well as the TaskEvent and displayed in the span view
- When resolving batchTriggerAndWait, the runtimes no longer reject promises, leading to an error in the parent task
apps/webapp/app/components/runs/RunFilters.tsx
+4 -6
View File
@@ -32,6 +32,7 @@ import {
import { TimeFrameFilter } from "./TimeFrameFilter";
import { Button } from "../primitives/Buttons";
import { useCallback } from "react";
import assertNever from "assert-never";
export function RunsFilters() {
const navigate = useNavigate();
@@ -182,8 +183,7 @@ export function FilterStatusIcon({
case "FAILED":
return <XCircleIcon className={cn(filterStatusClassNameColor(status), className)} />;
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
@@ -205,8 +205,7 @@ export function filterStatusTitle(status: FilterableStatus): string {
case "TIMEDOUT":
return "Timed out";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
@@ -228,8 +227,7 @@ export function filterStatusClassNameColor(status: FilterableStatus): string {
case "TIMEDOUT":
return "text-amber-300";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
apps/webapp/app/components/runs/RunStatuses.tsx
+4 -6
View File
@@ -11,6 +11,7 @@ import type { JobRunStatus } from "@trigger.dev/database";
import { cn } from "~/utils/cn";
import { Spinner } from "../primitives/Spinner";
import { z } from "zod";
import assertNever from "assert-never";
export function RunStatus({ status }: { status: JobRunStatus }) {
return (
@@ -51,8 +52,7 @@ export function RunStatusIcon({ status, className }: { status: JobRunStatus; cla
case "CANCELED":
return <NoSymbolIcon className={cn(runStatusClassNameColor(status), className)} />;
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
@@ -89,8 +89,7 @@ export function runStatusTitle(status: JobRunStatus): string {
case "INVALID_PAYLOAD":
return "Invalid payload";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
@@ -123,8 +122,7 @@ export function runStatusClassNameColor(status: JobRunStatus): string {
case "CANCELED":
return "text-charcoal-500";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
apps/webapp/app/components/runs/v3/DeploymentStatus.tsx
+4 -6
View File
@@ -5,6 +5,7 @@ import {
XCircleIcon,
} from "@heroicons/react/20/solid";
import { WorkerDeploymentStatus } from "@trigger.dev/database";
import assertNever from "assert-never";
import { Spinner } from "~/components/primitives/Spinner";
import { cn } from "~/utils/cn";
@@ -54,8 +55,7 @@ export function DeploymentStatusIcon({
/>
);
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
@@ -74,8 +74,7 @@ export function deploymentStatusClassNameColor(status: WorkerDeploymentStatus):
case "FAILED":
return "text-error";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
@@ -97,8 +96,7 @@ export function deploymentStatusTitle(status: WorkerDeploymentStatus): string {
case "FAILED":
return "Failed";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
apps/webapp/app/components/runs/v3/TaskRunAttemptStatus.tsx
+4 -6
View File
@@ -8,6 +8,7 @@ import {
} from "@heroicons/react/20/solid";
import type { TaskRunAttemptStatus as TaskRunAttemptStatusType } from "@trigger.dev/database";
import { TaskRunAttemptStatus } from "@trigger.dev/database";
import assertNever from "assert-never";
import { SnowflakeIcon } from "lucide-react";
import { Spinner } from "~/components/primitives/Spinner";
import { cn } from "~/utils/cn";
@@ -72,8 +73,7 @@ export function TaskRunAttemptStatusIcon({
case "COMPLETED":
return <CheckCircleIcon className={cn(runAttemptStatusClassNameColor(status), className)} />;
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
@@ -99,8 +99,7 @@ export function runAttemptStatusClassNameColor(status: ExtendedTaskAttemptStatus
case "COMPLETED":
return "text-success";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
@@ -126,8 +125,7 @@ export function runAttemptStatusTitle(status: ExtendedTaskAttemptStatus | null):
case "COMPLETED":
return "Completed";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
apps/webapp/app/components/runs/v3/TaskRunStatus.tsx
+4 -6
View File
@@ -10,6 +10,7 @@ import {
XCircleIcon,
} from "@heroicons/react/20/solid";
import { TaskRunStatus } from "@trigger.dev/database";
import assertNever from "assert-never";
import { SnowflakeIcon } from "lucide-react";
import { Spinner } from "~/components/primitives/Spinner";
import { cn } from "~/utils/cn";
@@ -88,8 +89,7 @@ export function TaskRunStatusIcon({
return <FireIcon className={cn(runStatusClassNameColor(status), className)} />;
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
@@ -120,8 +120,7 @@ export function runStatusClassNameColor(status: TaskRunStatus): string {
case "CRASHED":
return "text-error";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
@@ -153,8 +152,7 @@ export function runStatusTitle(status: TaskRunStatus): string {
case "CRASHED":
return "Crashed";
default: {
const _exhaustiveCheck: never = status;
throw new Error(`Non-exhaustive match for value: ${status}`);
assertNever(status);
}
}
}
apps/webapp/app/env.server.ts
-1
View File
@@ -76,7 +76,6 @@ const EnvironmentSchema = z.object({
REDIS_PASSWORD: z.string().optional(),
REDIS_TLS_DISABLED: z.string().optional(),
DEFAULT_QUEUE_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(5),
DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1),
apps/webapp/app/models/taskRun.server.ts
+128
View File
@@ -0,0 +1,128 @@
import {
TaskRunError,
TaskRunExecutionResult,
TaskRunFailedExecutionResult,
TaskRunSuccessfulExecutionResult,
} from "@trigger.dev/core/v3";
import {
BatchTaskRunItemStatus,
TaskRun,
TaskRunAttempt,
TaskRunAttemptStatus,
TaskRunStatus,
} from "@trigger.dev/database";
import { assertNever } from "assert-never";
import { logger } from "~/services/logger.server";
const SUCCESSFUL_STATUSES = [TaskRunStatus.COMPLETED_SUCCESSFULLY];
const FAILURE_STATUSES = [
TaskRunStatus.CANCELED,
TaskRunStatus.INTERRUPTED,
TaskRunStatus.COMPLETED_WITH_ERRORS,
TaskRunStatus.SYSTEM_FAILURE,
TaskRunStatus.CRASHED,
];
export type TaskRunWithAttempts = TaskRun & {
attempts: TaskRunAttempt[];
};
export function executionResultForTaskRun(
taskRun: TaskRunWithAttempts
): TaskRunExecutionResult | undefined {
if (SUCCESSFUL_STATUSES.includes(taskRun.status)) {
// find the last attempt that was successful
const attempt = taskRun.attempts.find((a) => a.status === TaskRunAttemptStatus.COMPLETED);
if (!attempt) {
logger.error("Task run is successful but no successful attempt found", {
taskRunId: taskRun.id,
taskRunStatus: taskRun.status,
taskRunAttempts: taskRun.attempts.map((a) => a.status),
});
return undefined;
}
return {
ok: true,
id: taskRun.friendlyId,
output: attempt.output ?? undefined,
outputType: attempt.outputType,
} satisfies TaskRunSuccessfulExecutionResult;
}
if (FAILURE_STATUSES.includes(taskRun.status)) {
if (taskRun.status === TaskRunStatus.CANCELED) {
return {
ok: false,
id: taskRun.friendlyId,
error: {
type: "INTERNAL_ERROR",
code: "TASK_RUN_CANCELLED",
},
} satisfies TaskRunFailedExecutionResult;
}
const attempt = taskRun.attempts.find((a) => a.status === TaskRunAttemptStatus.FAILED);
if (!attempt) {
logger.error("Task run is failed but no failed attempt found", {
taskRunId: taskRun.id,
taskRunStatus: taskRun.status,
taskRunAttempts: taskRun.attempts.map((a) => a.status),
});
return undefined;
}
const error = TaskRunError.safeParse(attempt.error);
if (!error.success) {
logger.error("Failed to parse error from failed task run attempt", {
taskRunId: taskRun.id,
taskRunStatus: taskRun.status,
taskRunAttempts: taskRun.attempts.map((a) => a.status),
error: attempt.error,
});
return {
ok: false,
id: taskRun.friendlyId,
error: {
type: "INTERNAL_ERROR",
code: "CONFIGURED_INCORRECTLY",
},
} satisfies TaskRunFailedExecutionResult;
}
return {
ok: false,
id: taskRun.friendlyId,
error: error.data,
} satisfies TaskRunFailedExecutionResult;
}
}
export function batchTaskRunItemStatusForRunStatus(status: TaskRunStatus): BatchTaskRunItemStatus {
switch (status) {
case TaskRunStatus.COMPLETED_SUCCESSFULLY:
return BatchTaskRunItemStatus.COMPLETED;
case TaskRunStatus.CANCELED:
case TaskRunStatus.INTERRUPTED:
case TaskRunStatus.COMPLETED_WITH_ERRORS:
case TaskRunStatus.SYSTEM_FAILURE:
case TaskRunStatus.CRASHED:
case TaskRunStatus.COMPLETED_WITH_ERRORS:
return BatchTaskRunItemStatus.FAILED;
case TaskRunStatus.PENDING:
case TaskRunStatus.WAITING_FOR_DEPLOY:
case TaskRunStatus.WAITING_TO_RESUME:
case TaskRunStatus.RETRYING_AFTER_FAILURE:
case TaskRunStatus.EXECUTING:
case TaskRunStatus.PAUSED:
return BatchTaskRunItemStatus.PENDING;
default:
assertNever(status);
}
}
apps/webapp/app/presenters/v3/ApiBatchResultsPresenter.server.ts
+46
View File
@@ -0,0 +1,46 @@
import { BatchTaskRunExecutionResult } from "@trigger.dev/core/v3";
import { executionResultForTaskRun } from "~/models/taskRun.server";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { BasePresenter } from "./basePresenter.server";
export class ApiBatchResultsPresenter extends BasePresenter {
public async call(
friendlyId: string,
env: AuthenticatedEnvironment
): Promise<BatchTaskRunExecutionResult | undefined> {
return this.traceWithEnv("call", env, async (span) => {
const batchRun = await this._prisma.batchTaskRun.findUnique({
where: {
friendlyId,
runtimeEnvironmentId: env.id,
},
include: {
items: {
include: {
taskRun: {
include: {
attempts: {
orderBy: {
createdAt: "desc",
},
},
},
},
},
},
},
});
if (!batchRun) {
return undefined;
}
return {
id: batchRun.friendlyId,
items: batchRun.items
.map((item) => executionResultForTaskRun(item.taskRun))
.filter(Boolean),
};
});
}
}
apps/webapp/app/presenters/v3/ApiRunResultPresenter.server.ts
+33
View File
@@ -0,0 +1,33 @@
import { TaskRunExecutionResult } from "@trigger.dev/core/v3";
import { executionResultForTaskRun } from "~/models/taskRun.server";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { BasePresenter } from "./basePresenter.server";
export class ApiRunResultPresenter extends BasePresenter {
public async call(
friendlyId: string,
env: AuthenticatedEnvironment
): Promise<TaskRunExecutionResult | undefined> {
return this.traceWithEnv("call", env, async (span) => {
const taskRun = await this._prisma.taskRun.findUnique({
where: {
friendlyId,
runtimeEnvironmentId: env.id,
},
include: {
attempts: {
orderBy: {
createdAt: "desc",
},
},
},
});
if (!taskRun) {
return undefined;
}
return executionResultForTaskRun(taskRun);
});
}
}
apps/webapp/app/presenters/v3/basePresenter.server.ts
+34
View File
@@ -0,0 +1,34 @@
import { Span, SpanKind } from "@opentelemetry/api";
import { PrismaClientOrTransaction, prisma } from "~/db.server";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { attributesFromAuthenticatedEnv, tracer } from "../../v3/tracer.server";
export abstract class BasePresenter {
constructor(protected readonly _prisma: PrismaClientOrTransaction = prisma) {}
protected async traceWithEnv<T>(
trace: string,
env: AuthenticatedEnvironment,
fn: (span: Span) => Promise<T>
): Promise<T> {
return tracer.startActiveSpan(
`${this.constructor.name}.${trace}`,
{ attributes: attributesFromAuthenticatedEnv(env), kind: SpanKind.SERVER },
async (span) => {
try {
return await fn(span);
} catch (e) {
if (e instanceof Error) {
span.recordException(e);
} else {
span.recordException(new Error(String(e)));
}
throw e;
} finally {
span.end();
}
}
);
}
}
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx
+3
View File
@@ -130,6 +130,9 @@ export default function Page() {
)}
<Property label="Message">{event.message}</Property>
<Property label="Task ID">{event.taskSlug}</Property>
{event.idempotencyKey && (
<Property label="Idempotency key">{event.idempotencyKey}</Property>
)}
{event.taskPath && event.taskExportName && (
<Property label="Task">
<TaskPath
apps/webapp/app/routes/api.v1.batches.$batchParam.results.ts
+45
View File
@@ -0,0 +1,45 @@
import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
import { json } from "@remix-run/server-runtime";
import { z } from "zod";
import { ApiBatchResultsPresenter } from "~/presenters/v3/ApiBatchResultsPresenter.server";
import { ApiRunResultPresenter } from "~/presenters/v3/ApiRunResultPresenter.server";
import { authenticateApiRequest } from "~/services/apiAuth.server";
const ParamsSchema = z.object({
/* This is the batch friendly ID */
batchParam: z.string(),
});
export async function loader({ request, params }: LoaderFunctionArgs) {
// Authenticate the request
const authenticationResult = await authenticateApiRequest(request);
if (!authenticationResult) {
return json({ error: "Invalid or Missing API Key" }, { status: 401 });
}
const parsed = ParamsSchema.safeParse(params);
if (!parsed.success) {
return json({ error: "Invalid or missing run ID" }, { status: 400 });
}
const { batchParam } = parsed.data;
try {
const presenter = new ApiBatchResultsPresenter();
const result = await presenter.call(batchParam, authenticationResult.environment);
if (!result) {
return json({ error: "Batch not found" }, { status: 404 });
}
return json(result);
} catch (error) {
if (error instanceof Error) {
return json({ error: error.message }, { status: 500 });
} else {
return json({ error: JSON.stringify(error) }, { status: 500 });
}
}
}
apps/webapp/app/routes/api.v1.runs.$runParam.replay.ts
+2 -4
View File
@@ -1,12 +1,10 @@
import type { ActionFunctionArgs } from "@remix-run/server-runtime";
import { json } from "@remix-run/server-runtime";
import { PrismaErrorSchema, prisma } from "~/db.server";
import { z } from "zod";
import { prisma } from "~/db.server";
import { authenticateApiRequest } from "~/services/apiAuth.server";
import { CancelRunService } from "~/services/runs/cancelRun.server";
import { ApiRunPresenter } from "~/presenters/ApiRunPresenter.server";
import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
import { logger } from "~/services/logger.server";
import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
const ParamsSchema = z.object({
/* This is the run friendly ID */
apps/webapp/app/routes/api.v1.runs.$runParam.result.ts
+44
View File
@@ -0,0 +1,44 @@
import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
import { json } from "@remix-run/server-runtime";
import { z } from "zod";
import { ApiRunResultPresenter } from "~/presenters/v3/ApiRunResultPresenter.server";
import { authenticateApiRequest } from "~/services/apiAuth.server";
const ParamsSchema = z.object({
/* This is the run friendly ID */
runParam: z.string(),
});
export async function loader({ request, params }: LoaderFunctionArgs) {
// Authenticate the request
const authenticationResult = await authenticateApiRequest(request);
if (!authenticationResult) {
return json({ error: "Invalid or Missing API Key" }, { status: 401 });
}
const parsed = ParamsSchema.safeParse(params);
if (!parsed.success) {
return json({ error: "Invalid or missing run ID" }, { status: 400 });
}
const { runParam } = parsed.data;
try {
const presenter = new ApiRunResultPresenter();
const result = await presenter.call(runParam, authenticationResult.environment);
if (!result) {
return json({ error: "Run either doesn't exist or is not finished" }, { status: 404 });
}
return json(result);
} catch (error) {
if (error instanceof Error) {
return json({ error: error.message }, { status: 500 });
} else {
return json({ error: JSON.stringify(error) }, { status: 500 });
}
}
}
apps/webapp/app/v3/eventRepository.server.ts
+4 -1
View File
@@ -63,6 +63,7 @@ export type TraceAttributes = Partial<
| "batchId"
| "payload"
| "payloadType"
| "idempotencyKey"
>
>;
@@ -371,6 +372,7 @@ export class EventRepository {
id: event.spanId,
parentId: event.parentId ?? undefined,
runId: event.runId,
idempotencyKey: event.idempotencyKey,
data: {
message: event.message,
style: event.style,
@@ -459,7 +461,7 @@ export class EventRepository {
const links: SpanLink[] = [];
if (messagingEvent.success && messagingEvent.data) {
if ("id" in messagingEvent.data.message) {
if (messagingEvent.data.message && "id" in messagingEvent.data.message) {
if (messagingEvent.data.message.id.startsWith("run_")) {
links.push({
type: "run",
@@ -719,6 +721,7 @@ export class EventRepository {
links: links as unknown as Prisma.InputJsonValue,
payload: options.attributes.payload,
payloadType: options.attributes.payloadType,
idempotencyKey: options.attributes.idempotencyKey,
};
if (options.immediate) {
apps/webapp/app/v3/marqs/devQueueConsumer.server.ts
+7 -5
View File
@@ -118,7 +118,7 @@ export class DevQueueConsumer {
completion: TaskRunExecutionResult,
execution: TaskRunExecution
) {
this._inProgressAttempts.delete(completion.id);
this._inProgressAttempts.delete(execution.attempt.id);
if (completion.ok) {
this._taskSuccesses++;
@@ -424,7 +424,7 @@ export class DevQueueConsumer {
orderBy: { number: "desc" },
},
tags: true,
batchItem: {
batchItems: {
include: {
batchTaskRun: true,
},
@@ -499,6 +499,7 @@ export class DevQueueConsumer {
createdAt: lockedTaskRun.createdAt,
tags: lockedTaskRun.tags.map((tag) => tag.name),
isTest: lockedTaskRun.isTest,
idempotencyKey: lockedTaskRun.idempotencyKey ?? undefined,
},
queue: {
id: queue.friendlyId,
@@ -520,9 +521,10 @@ export class DevQueueConsumer {
slug: this.env.project.slug,
name: this.env.project.name,
},
batch: lockedTaskRun.batchItem?.batchTaskRun
? { id: lockedTaskRun.batchItem.batchTaskRun.friendlyId }
: undefined,
batch:
lockedTaskRun.batchItems[0] && lockedTaskRun.batchItems[0].batchTaskRun
? { id: lockedTaskRun.batchItems[0].batchTaskRun.friendlyId }
: undefined,
};
const environmentRepository = new EnvironmentVariablesRepository();
apps/webapp/app/v3/marqs/index.server.ts
+20 -20
View File
@@ -39,7 +39,6 @@ const SemanticAttributes = {
export type MarQSOptions = {
redis: RedisOptions;
defaultQueueConcurrency: number;
defaultEnvConcurrency: number;
defaultOrgConcurrency: number;
windowSize?: number;
@@ -92,7 +91,7 @@ export class MarQS {
public async getQueueConcurrencyLimit(env: AuthenticatedEnvironment, queue: string) {
const result = await this.redis.get(this.keys.queueConcurrencyLimitKey(env, queue));
return result ? Number(result) : this.options.defaultQueueConcurrency;
return result ? Number(result) : undefined;
}
public async getEnvConcurrencyLimit(env: AuthenticatedEnvironment) {
@@ -860,7 +859,6 @@ export class MarQS {
messageQueue,
String(this.options.visibilityTimeoutInMs ?? 300000), // 5 minutes
String(Date.now()),
String(this.options.defaultQueueConcurrency),
String(this.options.defaultEnvConcurrency),
String(this.options.defaultOrgConcurrency)
);
@@ -1015,16 +1013,22 @@ export class MarQS {
concurrencyLimitKey,
envConcurrencyLimitKey,
orgConcurrencyLimitKey,
String(this.options.defaultQueueConcurrency),
String(this.options.defaultEnvConcurrency),
String(this.options.defaultOrgConcurrency)
);
const queueCurrent = Number(capacities[0]);
const envLimit = Number(capacities[3]);
const orgLimit = Number(capacities[5]);
const queueLimit = capacities[1] ? Number(capacities[1]) : Math.min(envLimit, orgLimit);
const envCurrent = Number(capacities[2]);
const orgCurrent = Number(capacities[4]);
// [queue current, queue limit, env current, env limit, org current, org limit]
return {
queue: { current: Number(capacities[0]), limit: Number(capacities[1]) },
env: { current: Number(capacities[2]), limit: Number(capacities[3]) },
org: { current: Number(capacities[4]), limit: Number(capacities[5]) },
queue: { current: queueCurrent, limit: queueLimit },
env: { current: envCurrent, limit: envLimit },
org: { current: orgCurrent, limit: orgLimit },
};
}
@@ -1119,13 +1123,12 @@ local currentConcurrencyKey = KEYS[7]
local envCurrentConcurrencyKey = KEYS[8]
local orgCurrentConcurrencyKey = KEYS[9]
-- Args: childQueueName, visibilityQueue, currentTime, defaultConcurrencyLimit, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
-- Args: childQueueName, visibilityQueue, currentTime, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
local childQueueName = ARGV[1]
local visibilityTimeout = tonumber(ARGV[2])
local currentTime = tonumber(ARGV[3])
local defaultConcurrencyLimit = ARGV[4]
local defaultEnvConcurrencyLimit = ARGV[5]
local defaultOrgConcurrencyLimit = ARGV[6]
local defaultEnvConcurrencyLimit = ARGV[4]
local defaultOrgConcurrencyLimit = ARGV[5]
-- Check current org concurrency against the limit
local orgCurrentConcurrency = tonumber(redis.call('SCARD', orgCurrentConcurrencyKey) or '0')
@@ -1145,8 +1148,9 @@ end
-- Check current queue concurrency against the limit
local currentConcurrency = tonumber(redis.call('SCARD', currentConcurrencyKey) or '0')
local concurrencyLimit = tonumber(redis.call('GET', concurrencyLimitKey) or defaultConcurrencyLimit)
local concurrencyLimit = tonumber(redis.call('GET', concurrencyLimitKey) or '1000000')
-- Check condition only if concurrencyLimit exists
if currentConcurrency >= concurrencyLimit then
return nil
end
@@ -1304,10 +1308,9 @@ local concurrencyLimitKey = KEYS[4]
local envConcurrencyLimitKey = KEYS[5]
local orgConcurrencyLimitKey = KEYS[6]
-- Args defaultConcurrencyLimit, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
local defaultConcurrencyLimit = tonumber(ARGV[1])
local defaultEnvConcurrencyLimit = tonumber(ARGV[2])
local defaultOrgConcurrencyLimit = tonumber(ARGV[3])
-- Args defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
local defaultEnvConcurrencyLimit = tonumber(ARGV[1])
local defaultOrgConcurrencyLimit = tonumber(ARGV[2])
local currentOrgConcurrency = tonumber(redis.call('SCARD', currentOrgConcurrencyKey) or '0')
local orgConcurrencyLimit = tonumber(redis.call('GET', orgConcurrencyLimitKey) or defaultOrgConcurrencyLimit)
@@ -1316,7 +1319,7 @@ local currentEnvConcurrency = tonumber(redis.call('SCARD', currentEnvConcurrency
local envConcurrencyLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
local currentConcurrency = tonumber(redis.call('SCARD', currentConcurrencyKey) or '0')
local concurrencyLimit = tonumber(redis.call('GET', concurrencyLimitKey) or defaultConcurrencyLimit)
local concurrencyLimit = redis.call('GET', concurrencyLimitKey)
-- Return current capacity and concurrency limits for the queue, env, org
return { currentConcurrency, concurrencyLimit, currentEnvConcurrency, envConcurrencyLimit, currentOrgConcurrency, orgConcurrencyLimit }
@@ -1398,7 +1401,6 @@ declare module "ioredis" {
childQueueName: string,
visibilityTimeout: string,
currentTime: string,
defaultConcurrencyLimit: string,
defaultEnvConcurrencyLimit: string,
defaultOrgConcurrencyLimit: string,
callback?: Callback<[string, string]>
@@ -1447,7 +1449,6 @@ declare module "ioredis" {
concurrencyLimitKey: string,
envConcurrencyLimitKey: string,
orgConcurrencyLimitKey: string,
defaultConcurrencyLimit: string,
defaultEnvConcurrencyLimit: string,
defaultOrgConcurrencyLimit: string,
callback?: Callback<number[]>
@@ -1492,7 +1493,6 @@ function getMarQSClient() {
envQueuePriorityStrategy: new SimpleWeightedChoiceStrategy({ queueSelectionCount: 12 }),
workers: 1,
redis: redisOptions,
defaultQueueConcurrency: env.DEFAULT_QUEUE_EXECUTION_CONCURRENCY_LIMIT,
defaultEnvConcurrency: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT,
defaultOrgConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
visibilityTimeoutInMs: 120 * 1000, // 2 minutes,
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
+8 -6
View File
@@ -812,7 +812,7 @@ class SharedQueueTasks {
if (ok) {
const success: TaskRunSuccessfulExecutionResult = {
ok,
id: attempt.friendlyId,
id: attempt.taskRun.friendlyId,
output: attempt.output ?? undefined,
outputType: attempt.outputType,
};
@@ -820,7 +820,7 @@ class SharedQueueTasks {
} else {
const failure: TaskRunFailedExecutionResult = {
ok,
id: attempt.friendlyId,
id: attempt.taskRun.friendlyId,
error: attempt.error as TaskRunError,
};
return failure;
@@ -848,7 +848,7 @@ class SharedQueueTasks {
taskRun: {
include: {
tags: true,
batchItem: {
batchItems: {
include: {
batchTaskRun: true,
},
@@ -956,6 +956,7 @@ class SharedQueueTasks {
createdAt: taskRun.createdAt,
tags: taskRun.tags.map((tag) => tag.name),
isTest: taskRun.isTest,
idempotencyKey: taskRun.idempotencyKey ?? undefined,
},
queue: {
id: queue.friendlyId,
@@ -977,9 +978,10 @@ class SharedQueueTasks {
slug: attempt.runtimeEnvironment.project.slug,
name: attempt.runtimeEnvironment.project.name,
},
batch: taskRun.batchItem?.batchTaskRun
? { id: taskRun.batchItem.batchTaskRun.friendlyId }
: undefined,
batch:
taskRun.batchItems[0] && taskRun.batchItems[0].batchTaskRun
? { id: taskRun.batchItems[0].batchTaskRun.friendlyId }
: undefined,
worker: {
id: attempt.backgroundWorkerId,
contentHash: attempt.backgroundWorker.contentHash,
apps/webapp/app/v3/marqs/types.ts
-1
View File
@@ -1,4 +1,3 @@
import { RedisOptions } from "ioredis";
import { z } from "zod";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
apps/webapp/app/v3/otlpExporter.server.ts
+1
View File
@@ -352,6 +352,7 @@ function extractResourceProperties(attributes: KeyValue[]) {
queueId: extractStringAttribute(attributes, SemanticInternalAttributes.QUEUE_ID),
queueName: extractStringAttribute(attributes, SemanticInternalAttributes.QUEUE_NAME),
batchId: extractStringAttribute(attributes, SemanticInternalAttributes.BATCH_ID),
idempotencyKey: extractStringAttribute(attributes, SemanticInternalAttributes.IDEMPOTENCY_KEY),
};
}
apps/webapp/app/v3/services/batchTriggerTask.server.ts
+19 -21
View File
@@ -1,9 +1,9 @@
import { BatchTriggerTaskRequestBody } from "@trigger.dev/core/v3";
import { nanoid } from "nanoid";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { generateFriendlyId } from "../friendlyIdentifiers";
import { BaseService } from "./baseService.server";
import { TriggerTaskService } from "./triggerTask.server";
import { batchTaskRunItemStatusForRunStatus } from "~/models/taskRun.server";
export type BatchTriggerTaskServiceOptions = {
idempotencyKey?: string;
@@ -22,23 +22,23 @@ export class BatchTriggerTaskService extends BaseService {
return await this.traceWithEnv("call()", environment, async (span) => {
span.setAttribute("taskId", taskId);
const idempotencyKey = options.idempotencyKey ?? nanoid();
const existingBatch = await this._prisma.batchTaskRun.findUnique({
where: {
runtimeEnvironmentId_idempotencyKey: {
runtimeEnvironmentId: environment.id,
idempotencyKey,
},
},
include: {
items: {
include: {
taskRun: true,
const existingBatch = options.idempotencyKey
? await this._prisma.batchTaskRun.findUnique({
where: {
runtimeEnvironmentId_idempotencyKey: {
runtimeEnvironmentId: environment.id,
idempotencyKey: options.idempotencyKey,
},
},
},
},
});
include: {
items: {
include: {
taskRun: true,
},
},
},
})
: undefined;
if (existingBatch) {
span.setAttribute("batchId", existingBatch.friendlyId);
@@ -58,7 +58,7 @@ export class BatchTriggerTaskService extends BaseService {
data: {
friendlyId: generateFriendlyId("batch"),
runtimeEnvironmentId: environment.id,
idempotencyKey,
idempotencyKey: options.idempotencyKey,
taskIdentifier: taskId,
dependentTaskAttemptId: dependentAttempt?.id,
},
@@ -70,8 +70,6 @@ export class BatchTriggerTaskService extends BaseService {
let index = 0;
for (const item of body.items) {
const idempotencyKey = nanoid();
const run = await triggerTaskService.call(
taskId,
environment,
@@ -83,7 +81,6 @@ export class BatchTriggerTaskService extends BaseService {
},
},
{
idempotencyKey,
triggerVersion: options.triggerVersion,
traceContext: options.traceContext,
spanParentAsLink: options.spanParentAsLink,
@@ -96,6 +93,7 @@ export class BatchTriggerTaskService extends BaseService {
data: {
batchTaskRunId: batch.id,
taskRunId: run.id,
status: batchTaskRunItemStatusForRunStatus(run.status),
},
});
apps/webapp/app/v3/services/cancelTaskRun.server.ts
+2 -2
View File
@@ -4,9 +4,9 @@ import { marqs } from "~/v3/marqs/index.server";
import { devPubSub } from "../marqs/devPubSub.server";
import { BaseService } from "./baseService.server";
import { socketIo } from "../handleSocketIo.server";
import { assertUnreachable } from "../utils/asserts.server";
import { CancelAttemptService } from "./cancelAttempt.server";
import { logger } from "~/services/logger.server";
import assertNever from "assert-never";
export const CANCELLABLE_STATUSES: Array<TaskRunStatus> = [
"PENDING",
@@ -148,7 +148,7 @@ export class CancelTaskRunService extends BaseService {
break;
}
default: {
assertUnreachable(attempt.status);
assertNever(attempt.status);
}
}
}
apps/webapp/app/v3/services/completeAttempt.server.ts
+6 -4
View File
@@ -39,10 +39,12 @@ export class CompleteAttemptService extends BaseService {
env?: AuthenticatedEnvironment;
checkpoint?: CheckpointData;
}): Promise<"COMPLETED" | "RETRIED"> {
const taskRunAttempt = await findAttempt(this._prisma, completion.id);
const taskRunAttempt = await findAttempt(this._prisma, execution.attempt.id);
if (!taskRunAttempt) {
logger.error("[CompleteAttemptService] Task run attempt not found", { id: completion.id });
logger.error("[CompleteAttemptService] Task run attempt not found", {
id: execution.attempt.id,
});
// Update the task run to be failed
await this._prisma.taskRun.update({
@@ -76,7 +78,7 @@ export class CompleteAttemptService extends BaseService {
env?: AuthenticatedEnvironment
): Promise<"COMPLETED"> {
await this._prisma.taskRunAttempt.update({
where: { friendlyId: completion.id },
where: { id: taskRunAttempt.id },
data: {
status: "COMPLETED",
completedAt: new Date(),
@@ -144,7 +146,7 @@ export class CompleteAttemptService extends BaseService {
}
await this._prisma.taskRunAttempt.update({
where: { friendlyId: completion.id },
where: { id: taskRunAttempt.id },
data: {
status: "FAILED",
completedAt: new Date(),
apps/webapp/app/v3/services/resumeTaskRunDependencies.server.ts
+7 -5
View File
@@ -12,7 +12,7 @@ export class ResumeTaskRunDependenciesService extends BaseService {
include: {
taskRun: {
include: {
batchItem: true,
batchItems: true,
dependency: {
include: {
dependentAttempt: true,
@@ -34,14 +34,16 @@ export class ResumeTaskRunDependenciesService extends BaseService {
return;
}
const { batchItem, dependency } = taskAttempt.taskRun;
const { batchItems, dependency } = taskAttempt.taskRun;
if (!batchItem && !dependency) {
if (!batchItems.length && !dependency) {
return;
}
if (batchItem) {
await this.#resumeBatchItem(batchItem, taskAttempt);
if (batchItems.length) {
for (const batchItem of batchItems) {
await this.#resumeBatchItem(batchItem, taskAttempt);
}
return;
}
apps/webapp/app/v3/services/triggerTask.server.ts
+13 -10
View File
@@ -34,18 +34,20 @@ export class TriggerTaskService extends BaseService {
return await this.traceWithEnv("call()", environment, async (span) => {
span.setAttribute("taskId", taskId);
const idempotencyKey = options.idempotencyKey ?? body.options?.idempotencyKey ?? nanoid();
const idempotencyKey = options.idempotencyKey ?? body.options?.idempotencyKey;
const existingRun = await this._prisma.taskRun.findUnique({
where: {
runtimeEnvironmentId_idempotencyKey: {
runtimeEnvironmentId: environment.id,
idempotencyKey,
},
},
});
const existingRun = idempotencyKey
? await this._prisma.taskRun.findUnique({
where: {
runtimeEnvironmentId_idempotencyKey: {
runtimeEnvironmentId: environment.id,
idempotencyKey,
},
},
})
: undefined;
if (existingRun) {
if (existingRun && existingRun.taskIdentifier === taskId) {
span.setAttribute("runId", existingRun.friendlyId);
return existingRun;
}
@@ -68,6 +70,7 @@ export class TriggerTaskService extends BaseService {
},
runIsTest: body.options?.test ?? false,
batchId: options.batchId,
idempotencyKey,
},
incomplete: true,
immediate: true,
apps/webapp/app/v3/utils/asserts.server.ts
-3
View File
@@ -1,3 +0,0 @@
export function assertUnreachable(x: never): never {
throw new Error("Didn't expect to get here");
}
apps/webapp/package.json
+1
View File
@@ -96,6 +96,7 @@
"@uiw/react-codemirror": "^4.19.5",
"@upstash/ratelimit": "^1.0.1",
"@whatwg-node/fetch": "^0.9.14",
"assert-never": "^1.2.1",
"aws4fetch": "^1.0.18",
"class-variance-authority": "^0.5.2",
"clsx": "^1.2.1",
packages/cli-v3/src/workers/dev/worker-facade.ts
+2 -2
View File
@@ -117,7 +117,7 @@ const handler = new ZodMessageHandler({
execution,
result: {
ok: false,
id: execution.attempt.id,
id: execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.TASK_ALREADY_RUNNING,
@@ -139,7 +139,7 @@ const handler = new ZodMessageHandler({
execution,
result: {
ok: false,
id: execution.attempt.id,
id: execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.COULD_NOT_FIND_EXECUTOR,
packages/cli-v3/src/workers/prod/worker-facade.ts
+3 -3
View File
@@ -110,7 +110,7 @@ const zodIpc = new ZodIpcConnection({
execution,
result: {
ok: false,
id: execution.attempt.id,
id: execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.TASK_ALREADY_RUNNING,
@@ -131,7 +131,7 @@ const zodIpc = new ZodIpcConnection({
execution,
result: {
ok: false,
id: execution.attempt.id,
id: execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.COULD_NOT_FIND_EXECUTOR,
@@ -182,7 +182,7 @@ const zodIpc = new ZodIpcConnection({
execution: _execution,
result: {
ok: false,
id: _execution.attempt.id,
id: _execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.GRACEFUL_EXIT_TIMEOUT,
packages/core/src/v3/apiClient/index.ts
+37
View File
@@ -1,6 +1,7 @@
import { context, propagation } from "@opentelemetry/api";
import { ZodFetchOptions, zodfetch } from "../zodfetch";
import {
BatchTaskRunExecutionResult,
BatchTriggerTaskRequestBody,
BatchTriggerTaskResponse,
CanceledRunResponse,
@@ -11,6 +12,7 @@ import {
ListSchedulesResult,
ReplayRunResponse,
ScheduleObject,
TaskRunExecutionResult,
TriggerTaskRequestBody,
TriggerTaskResponse,
UpdateScheduleOptions,
@@ -18,6 +20,7 @@ import {
import { taskContextManager } from "../tasks/taskContextManager";
import { getEnvVar } from "../utils/getEnv";
import { SafeAsyncLocalStorage } from "../utils/safeAsyncLocalStorage";
import { APIError } from "../apiErrors";
export type TriggerOptions = {
spanParentAsLink?: boolean;
@@ -46,6 +49,40 @@ export class ApiClient {
this.baseUrl = baseUrl.replace(/\/$/, "");
}
async getRunResult(runId: string): Promise<TaskRunExecutionResult | undefined> {
try {
return await zodfetch(
TaskRunExecutionResult,
`${this.baseUrl}/api/v1/runs/${runId}/result`,
{
method: "GET",
headers: this.#getHeaders(false),
},
zodFetchOptions
);
} catch (error) {
if (error instanceof APIError) {
if (error.status === 404) {
return undefined;
}
}
throw error;
}
}
async getBatchResults(batchId: string): Promise<BatchTaskRunExecutionResult | undefined> {
return await zodfetch(
BatchTaskRunExecutionResult,
`${this.baseUrl}/api/v1/batches/${batchId}/results`,
{
method: "GET",
headers: this.#getHeaders(false),
},
zodFetchOptions
);
}
triggerTask(taskId: string, body: TriggerTaskRequestBody, options?: TriggerOptions) {
return zodfetch(
TriggerTaskResponse,
packages/core/src/v3/logger/taskLogger.ts
+13 -1
View File
@@ -108,8 +108,20 @@ export class NoopTaskLogger implements TaskLogger {
function safeJsonProcess(value?: Record<string, unknown>): Record<string, unknown> | undefined {
try {
return JSON.parse(JSON.stringify(value));
return JSON.parse(JSON.stringify(value, jsonErrorReplacer));
} catch {
return value;
}
}
function jsonErrorReplacer(key: string, value: unknown) {
if (value instanceof Error) {
return {
name: value.name,
message: value.message,
stack: value.stack,
};
}
return value;
}
packages/core/src/v3/runtime/devRuntimeManager.ts
+9 -9
View File
@@ -10,7 +10,7 @@ import { unboundedTimeout } from "../utils/timers";
export class DevRuntimeManager implements RuntimeManager {
_taskWaits: Map<
string,
{ resolve: (value: TaskRunExecutionResult) => void; reject: (err?: any) => void }
{ resolve: (value: TaskRunExecutionResult) => void; reject?: (err?: any) => void }
> = new Map();
_batchWaits: Map<
@@ -65,16 +65,12 @@ export class DevRuntimeManager implements RuntimeManager {
if (pendingCompletion) {
this._pendingCompletionNotifications.delete(runId);
if (pendingCompletion.ok) {
resolve(pendingCompletion);
} else {
reject(pendingCompletion);
}
resolve(pendingCompletion);
return;
}
this._taskWaits.set(runId, { resolve, reject });
this._taskWaits.set(runId, { resolve });
});
})
);
@@ -97,10 +93,14 @@ export class DevRuntimeManager implements RuntimeManager {
return;
}
if (completion.ok) {
if (!wait.reject) {
wait.resolve(completion);
} else {
wait.reject(completion);
if (completion.ok) {
wait.resolve(completion);
} else {
wait.reject(completion);
}
}
this._taskWaits.delete(execution.run.id);
packages/core/src/v3/runtime/prodRuntimeManager.ts
+8 -4
View File
@@ -19,7 +19,7 @@ export type ProdRuntimeManagerOptions = {
export class ProdRuntimeManager implements RuntimeManager {
_taskWaits: Map<
string,
{ resolve: (value: TaskRunExecutionResult) => void; reject: (err?: any) => void }
{ resolve: (value: TaskRunExecutionResult) => void; reject?: (err?: any) => void }
> = new Map();
_batchWaits: Map<
@@ -114,7 +114,7 @@ export class ProdRuntimeManager implements RuntimeManager {
const promise = Promise.all(
params.runs.map((runId) => {
return new Promise<TaskRunExecutionResult>((resolve, reject) => {
this._taskWaits.set(runId, { resolve, reject });
this._taskWaits.set(runId, { resolve });
});
})
);
@@ -139,10 +139,14 @@ export class ProdRuntimeManager implements RuntimeManager {
return;
}
if (completion.ok) {
if (!wait.reject) {
wait.resolve(completion);
} else {
wait.reject(completion);
if (completion.ok) {
wait.resolve(completion);
} else {
wait.reject(completion);
}
}
this._taskWaits.delete(execution.run.id);
packages/core/src/v3/schemas/common.ts
+1
View File
@@ -74,6 +74,7 @@ export const TaskRun = z.object({
tags: z.array(z.string()),
isTest: z.boolean().default(false),
createdAt: z.coerce.date(),
idempotencyKey: z.string().optional(),
});
export type TaskRun = z.infer<typeof TaskRun>;
packages/core/src/v3/semanticInternalAttributes.ts
+1
View File
@@ -41,4 +41,5 @@ export const SemanticInternalAttributes = {
RETRY_DELAY: "retry.delay",
RETRY_COUNT: "retry.count",
LINK_TITLE: "$link.title",
IDEMPOTENCY_KEY: "ctx.run.idempotencyKey",
};
packages/core/src/v3/tasks/taskContextManager.ts
+1
View File
@@ -68,6 +68,7 @@ export class TaskContextManager {
[SemanticInternalAttributes.ORGANIZATION_SLUG]: this.ctx.organization.slug,
[SemanticInternalAttributes.ORGANIZATION_NAME]: this.ctx.organization.name,
[SemanticInternalAttributes.BATCH_ID]: this.ctx.batch?.id,
[SemanticInternalAttributes.IDEMPOTENCY_KEY]: this.ctx.run.idempotencyKey,
};
}
packages/core/src/v3/workers/taskExecutor.ts
+4 -4
View File
@@ -116,7 +116,7 @@ export class TaskExecutor {
return {
ok: true,
id: execution.attempt.id,
id: execution.run.id,
output: finalOutput.data,
outputType: finalOutput.dataType,
} satisfies TaskRunExecutionResult;
@@ -125,7 +125,7 @@ export class TaskExecutor {
return {
ok: false,
id: execution.attempt.id,
id: execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.TASK_OUTPUT_ERROR,
@@ -150,7 +150,7 @@ export class TaskExecutor {
recordSpanException(span, handleErrorResult.error ?? runError);
return {
id: execution.attempt.id,
id: execution.run.id,
ok: false,
error: handleErrorResult.error
? parseError(handleErrorResult.error)
@@ -164,7 +164,7 @@ export class TaskExecutor {
return {
ok: false,
id: execution.attempt.id,
id: execution.run.id,
error: {
type: "INTERNAL_ERROR",
code: TaskRunErrorCodes.HANDLE_ERROR_ERROR,
packages/database/prisma/migrations/20240418092931_make_idempotency_key_optional/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "TaskRun" ALTER COLUMN "idempotencyKey" DROP NOT NULL;
packages/database/prisma/migrations/20240418093153_add_idempotency_key_to_task_event/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "TaskEvent" ADD COLUMN "idempotencyKey" TEXT;
packages/database/prisma/migrations/20240418100819_make_batch_task_run_idempotency_key_optional/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "BatchTaskRun" ALTER COLUMN "idempotencyKey" DROP NOT NULL;
packages/database/prisma/migrations/20240418114019_make_batch_run_item_run_id_non_unique/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- DropIndex
DROP INDEX "BatchTaskRunItem_taskRunId_key";
packages/database/prisma/schema.prisma
+6 -4
View File
@@ -1588,7 +1588,7 @@ model TaskRun {
status TaskRunStatus @default(PENDING)
idempotencyKey String
idempotencyKey String?
taskIdentifier String
isTest Boolean @default(false)
@@ -1625,7 +1625,7 @@ model TaskRun {
concurrencyKey String?
batchItem BatchTaskRunItem?
batchItems BatchTaskRunItem[]
dependency TaskRunDependency?
CheckpointRestoreEvent CheckpointRestoreEvent[]
@@ -1816,6 +1816,8 @@ model TaskEvent {
runId String
runIsTest Boolean @default(false)
idempotencyKey String?
taskSlug String
taskPath String?
taskExportName String?
@@ -1913,7 +1915,7 @@ model BatchTaskRun {
status BatchTaskRunStatus @default(PENDING)
idempotencyKey String
idempotencyKey String?
taskIdentifier String
checkpointEvent CheckpointRestoreEvent? @relation(fields: [checkpointEventId], references: [id], onDelete: Cascade, onUpdate: Cascade)
@@ -1948,7 +1950,7 @@ model BatchTaskRunItem {
batchTaskRunId String
taskRun TaskRun @relation(fields: [taskRunId], references: [id], onDelete: Cascade, onUpdate: Cascade)
taskRunId String @unique
taskRunId String
taskRunAttempt TaskRunAttempt? @relation(fields: [taskRunAttemptId], references: [id], onDelete: SetNull, onUpdate: Cascade)
taskRunAttemptId String?
packages/trigger-sdk/src/v3/shared.ts
+82 -2
View File
@@ -5,6 +5,7 @@ import {
SEMATTRS_MESSAGING_SYSTEM,
} from "@opentelemetry/semantic-conventions";
import {
BatchTaskRunExecutionResult,
HandleErrorFnParams,
HandleErrorResult,
InitFnParams,
@@ -24,6 +25,7 @@ import {
conditionallyImportPacket,
createErrorTaskError,
defaultRetryOptions,
logger,
parsePacket,
runtime,
stringifyIO,
@@ -378,6 +380,29 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
span.setAttribute("messaging.message.id", response.id);
if (options?.idempotencyKey) {
// If an idempotency key is provided, we can check if the result is already available
const result = await apiClient.getRunResult(response.id);
if (result) {
logger.log(
`Result reused from previous task run with idempotency key '${options.idempotencyKey}'.`,
{
runId: response.id,
idempotencyKey: options.idempotencyKey,
}
);
const runResult = await handleTaskRunExecutionResult<TOutput>(result);
if (!runResult.ok) {
throw runResult.error;
}
return runResult.output;
}
}
const result = await runtime.waitForTask({
id: response.id,
ctx,
@@ -455,13 +480,68 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
span.setAttribute("messaging.message.id", response.batchId);
const getBatchResults = async (): Promise<BatchTaskRunExecutionResult> => {
// We need to check if the results are already available, but only if any of the items options has an idempotency key
const hasIdempotencyKey = items.some((item) => item.options?.idempotencyKey);
if (hasIdempotencyKey) {
const results = await apiClient.getBatchResults(response.batchId);
if (results) {
return results;
}
}
return {
id: response.batchId,
items: [],
};
};
const existingResults = await getBatchResults();
const incompleteRuns = response.runs.filter(
(runId) => !existingResults.items.some((item) => item.id === runId)
);
if (incompleteRuns.length === 0) {
logger.log(
`Results reused from previous task runs because of the provided idempotency keys.`
);
// All runs are already completed
const runs = await handleBatchTaskRunExecutionResult<TOutput>(existingResults.items);
return {
id: existingResults.id,
runs,
};
}
const result = await runtime.waitForBatch({
id: response.batchId,
runs: response.runs,
runs: incompleteRuns,
ctx,
});
const runs = await handleBatchTaskRunExecutionResult<TOutput>(result.items);
// Combine the already completed runs with the newly completed runs, ordered by the original order
const combinedItems: BatchTaskRunExecutionResult["items"] = [];
for (const runId of response.runs) {
const existingItem = existingResults.items.find((item) => item.id === runId);
if (existingItem) {
combinedItems.push(existingItem);
} else {
const newItem = result.items.find((item) => item.id === runId);
if (newItem) {
combinedItems.push(newItem);
}
}
}
const runs = await handleBatchTaskRunExecutionResult<TOutput>(combinedItems);
return {
id: result.id,
pnpm-lock.yaml
Generated
+7
View File
@@ -402,6 +402,9 @@ importers:
'@whatwg-node/fetch':
specifier: ^0.9.14
version: 0.9.14
assert-never:
specifier: ^1.2.1
version: 1.2.1
aws4fetch:
specifier: ^1.0.18
version: 1.0.18
@@ -15827,6 +15830,10 @@ packages:
tslib: 2.6.2
dev: false
/assert-never@1.2.1:
resolution: {integrity: sha512-TaTivMB6pYI1kXwrFlEhLeGfOqoDNdTxjCdwRfFFkEA30Eu+k48W34nlok2EYWJfFFzqaEmichdNM7th6M5HNw==}
dev: false
/assert-plus@1.0.0:
resolution: {integrity: sha512-NfJ4UzBCcQGLDlQq7nHxH+tv3kyZ0hHQqF5BO6J7tNJeP5do1llPr8dZ8zHonfhAu0PHAdMkSo+8o0wxg9lZWw==}
engines: {node: '>=0.8'}
references/v3-catalog/src/handleError.ts
+4 -2
View File
@@ -1,5 +1,7 @@
import type { HandleErrorFunction } from "@trigger.dev/sdk/v3";
import { logger, type HandleErrorFunction } from "@trigger.dev/sdk/v3";
export const handleError: HandleErrorFunction = async (payload, error, { ctx, retry }) => {
console.log("GOT TO handleError FUNCTION");
logger.log("handling error", { error });
return { skipRetrying: true };
};
references/v3-catalog/src/trigger/idempotencyKeys.ts
+78
View File
@@ -0,0 +1,78 @@
import { task, wait } from "@trigger.dev/sdk/v3";
export const idempotencyKeyParent = task({
id: "idempotency-key-parent",
run: async (payload: { key: string }) => {
console.log("Hello from idempotency-key-parent");
const childTaskResponse = await idempotencyKeyChild.triggerAndWait({
payload: {
key: payload.key,
forceError: true,
},
options: {
idempotencyKey: payload.key,
},
});
return {
key: payload.key,
childTaskResponse,
};
},
});
export const idempotencyKeyChild = task({
id: "idempotency-key-child",
run: async (payload: { forceError: boolean; key: string }) => {
console.log("Hello from idempotency-key-child", payload.key);
await wait.for({ seconds: 5 });
if (payload.forceError) {
throw new Error("This is a forced error in idempotency-key-child");
}
return payload;
},
});
export const idempotencyKeyBatchParent = task({
id: "idempotency-key-batch-parent",
run: async (payload: { keyPrefix: string; itemCount: number }) => {
console.log("Hello from idempotency-key-batch-parent");
const childTaskResponse = await idempotencyKeyBatchChild.batchTriggerAndWait({
items: Array.from({ length: payload.itemCount }).map((_, index) => ({
payload: {
key: `${payload.keyPrefix}-${index}`,
forceError: index % 2 === 0,
waitSeconds: 5 * index,
},
options: {
idempotencyKey: `${payload.keyPrefix}-${index}`,
},
})),
});
return {
keyPrefix: payload.keyPrefix,
childTaskResponse,
};
},
});
export const idempotencyKeyBatchChild = task({
id: "idempotency-key-batch-child",
run: async (payload: { forceError: boolean; key: string; waitSeconds: number }) => {
console.log("idempotency-key-batch-child", payload.key);
await wait.for({ seconds: payload.waitSeconds });
if (payload.forceError) {
throw new Error(`This is a forced error in idempotency-key-batch-child ${payload.key}`);
}
return payload;
},
});