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

feat: Run metadata (#1357)

Browse Source 
* Run metadata

* Remove metadata from context, move it to it’s own tab

* More run metadata stuff

- Add metadata to testing
- Make using metadata outside of runs a no-op
- Add docs

* Replaying should copy over the metadata

* transfer final attempt output to the task run

* A couple of minor fixes

* Use the new clientOrThrow() method everywhere

* Cleaned up the update metadata endpoint and added an API doc page for it

* Mirror task run attempt errors and output
This commit is contained in:
Eric Allam
2024-09-26 09:23:38 +01:00
committed by GitHub
parent d74783c231
commit 3c492bc202
48 changed files with 1292 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/tasty-rats-rhyme.md
+6
View File
@@ -0,0 +1,6 @@
---
"@trigger.dev/sdk": patch
"@trigger.dev/core": patch
---
Add Run metadata to allow for storing up to 4KB of data on a run and update it during the run
apps/webapp/app/components/code/JSONEditor.tsx
+2 -2
View File
@@ -112,9 +112,9 @@ export function JSONEditor(opts: JSONEditorProps) {
return (
<div
className={cn(
opts.className,
"grid",
showButtons ? "grid-rows-[2.5rem_1fr]" : "grid-rows-[1fr]"
showButtons ? "grid-rows-[2.5rem_1fr]" : "grid-rows-[1fr]",
opts.className
)}
>
{showButtons && (
apps/webapp/app/components/primitives/Tabs.tsx
+1
View File
@@ -97,6 +97,7 @@ export function TabButton({
return (
<button
className={cn("group flex flex-col items-center pt-1", props.className)}
type="button"
ref={ref}
{...props}
>
apps/webapp/app/env.server.ts
+1
View File
@@ -212,6 +212,7 @@ const EnvironmentSchema = z.object({
MAXIMUM_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000),
TASK_PAYLOAD_OFFLOAD_THRESHOLD: z.coerce.number().int().default(524_288), // 512KB
TASK_PAYLOAD_MAXIMUM_SIZE: z.coerce.number().int().default(3_145_728), // 3MB
TASK_RUN_METADATA_MAXIMUM_SIZE: z.coerce.number().int().default(4_096), // 4KB
});
export type Environment = z.infer<typeof EnvironmentSchema>;
apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts
+18 -8
View File
@@ -29,6 +29,8 @@ const commonRunSelect = {
completedAt: true,
expiredAt: true,
delayUntil: true,
metadata: true,
metadataType: true,
ttl: true,
tags: true,
costInCents: true,
@@ -157,10 +159,8 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
}
}
const apiStatus = ApiRetrieveRunPresenter.apiStatusFromRunStatus(taskRun.status);
return {
...createCommonRunStructure(taskRun),
...(await createCommonRunStructure(taskRun)),
payload: $payload,
payloadPresignedUrl: $payloadPresignedUrl,
output: $output,
@@ -191,11 +191,15 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
error: ApiRetrieveRunPresenter.apiErrorFromError(a.error),
})),
relatedRuns: {
root: taskRun.rootTaskRun ? createCommonRunStructure(taskRun.rootTaskRun) : undefined,
parent: taskRun.parentTaskRun
? createCommonRunStructure(taskRun.parentTaskRun)
root: taskRun.rootTaskRun
? await createCommonRunStructure(taskRun.rootTaskRun)
: undefined,
children: taskRun.childRuns.map((r) => createCommonRunStructure(r)),
parent: taskRun.parentTaskRun
? await createCommonRunStructure(taskRun.parentTaskRun)
: undefined,
children: await Promise.all(
taskRun.childRuns.map(async (r) => await createCommonRunStructure(r))
),
},
};
});
@@ -329,7 +333,12 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
}
}
function createCommonRunStructure(run: CommonRelatedRun) {
async function createCommonRunStructure(run: CommonRelatedRun) {
const metadata = await parsePacket({
data: run.metadata ?? undefined,
dataType: run.metadataType,
});
return {
id: run.friendlyId,
taskIdentifier: run.taskIdentifier,
@@ -354,6 +363,7 @@ function createCommonRunStructure(run: CommonRelatedRun) {
...ApiRetrieveRunPresenter.apiBooleanHelpersFromTaskRunStatus(run.status),
triggerFunction: resolveTriggerFunction(run),
batchId: run.batch?.friendlyId,
metadata,
};
}
apps/webapp/app/presenters/v3/SpanPresenter.server.ts
+13 -1
View File
@@ -1,4 +1,9 @@
import { MachinePresetName, prettyPrintPacket, TaskRunError } from "@trigger.dev/core/v3";
import {
MachinePresetName,
parsePacket,
prettyPrintPacket,
TaskRunError,
} from "@trigger.dev/core/v3";
import { RUNNING_STATUSES } from "~/components/runs/v3/TaskRunStatus";
import { eventRepository } from "~/v3/eventRepository.server";
import { machinePresetFromName } from "~/v3/machinePresets.server";
@@ -113,6 +118,8 @@ export class SpanPresenter extends BasePresenter {
},
payload: true,
payloadType: true,
metadata: true,
metadataType: true,
maxAttempts: true,
project: {
include: {
@@ -185,6 +192,10 @@ export class SpanPresenter extends BasePresenter {
const span = await eventRepository.getSpan(spanId, run.traceId);
const metadata = run.metadata
? await prettyPrintPacket(run.metadata, run.metadataType)
: undefined;
const context = {
task: {
id: run.taskIdentifier,
@@ -272,6 +283,7 @@ export class SpanPresenter extends BasePresenter {
error,
links: span?.links,
context: JSON.stringify(context, null, 2),
metadata,
};
}
apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts
+9 -7
View File
@@ -1,11 +1,6 @@
import { ScheduledTaskPayload, parsePacket, prettyPrintPacket } from "@trigger.dev/core/v3";
import {
RuntimeEnvironmentType,
TaskRunAttemptStatus,
TaskRunStatus,
TaskTriggerSource,
} from "@trigger.dev/database";
import { sqlDatabaseSchema, PrismaClient, prisma } from "~/db.server";
import { RuntimeEnvironmentType, TaskRunStatus } from "@trigger.dev/database";
import { PrismaClient, prisma, sqlDatabaseSchema } from "~/db.server";
import { getTimezones } from "~/utils/timezones.server";
import { getUsername } from "~/utils/username";
@@ -51,6 +46,8 @@ type RawRun = {
payload: string;
payloadType: string;
runtimeEnvironmentId: string;
metadata?: string;
metadataType?: string;
};
export type StandardRun = Omit<RawRun, "number"> & {
@@ -131,6 +128,8 @@ export class TestTaskPresenter {
taskr.status,
taskr.payload,
taskr."payloadType",
taskr.metadata,
taskr."metadataType",
taskr."runtimeEnvironmentId"
FROM
taskruns AS taskr
@@ -166,6 +165,9 @@ export class TestTaskPresenter {
...r,
number,
payload: await prettyPrintPacket(r.payload, r.payloadType),
metadata: r.metadata
? await prettyPrintPacket(r.metadata, r.metadataType)
: undefined,
};
})
),
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.test.tasks.$taskParam/route.tsx
+96 -30
View File
@@ -27,9 +27,11 @@ import {
ResizablePanelGroup,
} from "~/components/primitives/Resizable";
import { Select } from "~/components/primitives/Select";
import { TabButton, TabContainer } from "~/components/primitives/Tabs";
import { TextLink } from "~/components/primitives/TextLink";
import { TaskRunStatusCombo } from "~/components/runs/v3/TaskRunStatus";
import { TimezoneList } from "~/components/scheduled/timezones";
import { useSearchParams } from "~/hooks/useSearchParam";
import { redirectBackWithErrorMessage, redirectWithSuccessMessage } from "~/models/message.server";
import {
ScheduledRun,
@@ -39,6 +41,7 @@ import {
} from "~/presenters/v3/TestTaskPresenter.server";
import { logger } from "~/services/logger.server";
import { requireUserId } from "~/services/session.server";
import { cn } from "~/utils/cn";
import { docsPath, v3RunSpanPath, v3TaskParamsSchema } from "~/utils/pathBuilder";
import { TestTaskService } from "~/v3/services/testTask.server";
import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
@@ -129,27 +132,44 @@ export default function Page() {
const startingJson = "{\n\n}";
function StandardTaskForm({ task, runs }: { task: TestTask["task"]; runs: StandardRun[] }) {
const { value, replace } = useSearchParams();
const tab = value("tab");
//form submission
const submit = useSubmit();
const lastSubmission = useActionData();
//recent runs
const [selectedCodeSampleId, setSelectedCodeSampleId] = useState(runs.at(0)?.id);
const selectedCodeSample = runs.find((r) => r.id === selectedCodeSampleId)?.payload;
const selectedCodeSample = runs.find((r) => r.id === selectedCodeSampleId);
const selectedCodeSamplePayload = selectedCodeSample?.payload;
const selectedCodeSampleMetadata = selectedCodeSample?.metadata;
const [defaultJson, setDefaultJson] = useState<string>(selectedCodeSample ?? startingJson);
const setCode = useCallback((code: string) => {
setDefaultJson(code);
const [defaultPayloadJson, setDefaultPayloadJson] = useState<string>(
selectedCodeSamplePayload ?? startingJson
);
const setPayload = useCallback((code: string) => {
setDefaultPayloadJson(code);
}, []);
const currentJson = useRef<string>(defaultJson);
const currentPayloadJson = useRef<string>(defaultPayloadJson);
const [defaultMetadataJson, setDefaultMetadataJson] = useState<string>(
selectedCodeSampleMetadata ?? "{}"
);
const setMetadata = useCallback((code: string) => {
setDefaultMetadataJson(code);
}, []);
const currentMetadataJson = useRef<string>(defaultMetadataJson);
const submitForm = useCallback(
(e: React.FormEvent<HTMLFormElement>) => {
submit(
{
triggerSource: "STANDARD",
payload: currentJson.current,
payload: currentPayloadJson.current,
metadata: currentMetadataJson.current,
taskIdentifier: task.taskIdentifier,
environmentId: task.environment.id,
},
@@ -160,7 +180,7 @@ function StandardTaskForm({ task, runs }: { task: TestTask["task"]; runs: Standa
);
e.preventDefault();
},
[currentJson]
[currentPayloadJson, currentMetadataJson]
);
const [form, { environmentId, payload }] = useForm({
@@ -183,28 +203,73 @@ function StandardTaskForm({ task, runs }: { task: TestTask["task"]; runs: Standa
<ResizablePanelGroup orientation="horizontal">
<ResizablePanel id="test-task-main" min="100px" default="60%">
<div className="h-full bg-charcoal-900">
<JSONEditor
defaultValue={defaultJson}
readOnly={false}
basicSetup
onChange={(v) => {
currentJson.current = v;
<TabContainer className="px-3 pt-2">
<TabButton
isActive={!tab || tab === "payload"}
layoutId="test-editor"
onClick={() => {
replace({ tab: "payload" });
}}
>
Payload
</TabButton>
//deselect the example if it's been edited
if (selectedCodeSampleId) {
if (v !== selectedCodeSample) {
setDefaultJson(v);
setSelectedCodeSampleId(undefined);
<TabButton
isActive={tab === "metadata"}
layoutId="test-editor"
onClick={() => {
replace({ tab: "metadata" });
}}
>
Metadata
</TabButton>
</TabContainer>
<div>
<JSONEditor
defaultValue={defaultPayloadJson}
readOnly={false}
basicSetup
onChange={(v) => {
currentPayloadJson.current = v;
//deselect the example if it's been edited
if (selectedCodeSampleId) {
if (v !== selectedCodeSamplePayload) {
setDefaultPayloadJson(v);
setSelectedCodeSampleId(undefined);
}
}
}
}}
height="100%"
min-height="100%"
max-height="100%"
autoFocus
placeholder="Use your schema to enter valid JSON or add one of the recent payloads then click 'Run test'"
className="h-full"
/>
}}
height="100%"
min-height="100%"
max-height="100%"
autoFocus={!tab || tab === "payload"}
placeholder="{ }"
className={cn("h-full", tab === "metadata" && "hidden")}
/>
<JSONEditor
defaultValue={defaultMetadataJson}
readOnly={false}
basicSetup
onChange={(v) => {
currentMetadataJson.current = v;
//deselect the example if it's been edited
if (selectedCodeSampleId) {
if (v !== selectedCodeSampleMetadata) {
setDefaultMetadataJson(v);
setSelectedCodeSampleId(undefined);
}
}
}}
height="100%"
min-height="100%"
max-height="100%"
autoFocus={tab === "metadata"}
placeholder=""
className={cn("h-full", tab !== "metadata" && "hidden")}
/>
</div>
</div>
</ResizablePanel>
<ResizableHandle id="test-task-handle" />
@@ -213,9 +278,10 @@ function StandardTaskForm({ task, runs }: { task: TestTask["task"]; runs: Standa
runs={runs}
selectedId={selectedCodeSampleId}
onSelected={(id) => {
const payload = runs.find((r) => r.id === id)?.payload;
if (!payload) return;
setCode(payload);
const run = runs.find((r) => r.id === id);
if (!run) return;
setPayload(run.payload);
run.metadata && setMetadata(run.metadata);
setSelectedCodeSampleId(id);
}}
/>
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.test/route.tsx
+6 -1
View File
@@ -75,8 +75,13 @@ export default function Page() {
const navigation = useNavigation();
const location = useLocation();
const locationSearchParams = new URLSearchParams(location.search);
const navigationSearchParams = new URLSearchParams(navigation.location?.search);
const isLoadingTasks =
navigation.state === "loading" && navigation.location.pathname === location.pathname;
navigation.state === "loading" &&
navigation.location.pathname === location.pathname &&
navigationSearchParams.get("environment") !== locationSearchParams.get("environment");
return (
<PageContainer>
apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts
+94
View File
@@ -0,0 +1,94 @@
import { type ActionFunctionArgs, json } from "@remix-run/server-runtime";
import { parsePacket, UpdateMetadataRequestBody } from "@trigger.dev/core/v3";
import { z } from "zod";
import { prisma } from "~/db.server";
import { authenticateApiRequest } from "~/services/apiAuth.server";
import { handleMetadataPacket } from "~/utils/packets";
import { ServiceValidationError } from "~/v3/services/baseService.server";
import { isFinalRunStatus } from "~/v3/taskStatus";
const ParamsSchema = z.object({
runId: z.string(),
});
export async function action({ request, params }: ActionFunctionArgs) {
// Ensure this is a PUT request
if (request.method.toUpperCase() !== "PUT") {
return json({ error: "Method not allowed" }, { status: 405, headers: { Allow: "PUT" } });
}
// Authenticate the request
const authenticationResult = await authenticateApiRequest(request);
if (!authenticationResult) {
return json({ error: "Invalid or Missing API Key" }, { status: 401 });
}
const parsedParams = ParamsSchema.safeParse(params);
if (!parsedParams.success) {
return json(
{ error: "Invalid request parameters", issues: parsedParams.error.issues },
{ status: 400 }
);
}
try {
const anyBody = await request.json();
const body = UpdateMetadataRequestBody.safeParse(anyBody);
if (!body.success) {
return json({ error: "Invalid request body", issues: body.error.issues }, { status: 400 });
}
const metadataPacket = handleMetadataPacket(
body.data.metadata,
body.data.metadataType ?? "application/json"
);
if (!metadataPacket) {
return json({ error: "Invalid metadata" }, { status: 400 });
}
const taskRun = await prisma.taskRun.findFirst({
where: {
friendlyId: parsedParams.data.runId,
runtimeEnvironmentId: authenticationResult.environment.id,
},
select: {
status: true,
},
});
if (!taskRun) {
return json({ error: "Task Run not found" }, { status: 404 });
}
if (isFinalRunStatus(taskRun.status)) {
return json({ error: "Cannot update metadata for a completed run" }, { status: 400 });
}
await prisma.taskRun.update({
where: {
friendlyId: parsedParams.data.runId,
runtimeEnvironmentId: authenticationResult.environment.id,
},
data: {
metadata: metadataPacket?.data,
metadataType: metadataPacket?.dataType,
},
});
const parsedPacket = await parsePacket(metadataPacket);
return json({ metadata: parsedPacket }, { status: 200 });
} catch (error) {
if (error instanceof ServiceValidationError) {
return json({ error: error.message }, { status: error.status ?? 422 });
} else {
return json(
{ error: error instanceof Error ? error.message : "Internal Server Error" },
{ status: 500 }
);
}
}
}
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx
+21
View File
@@ -421,6 +421,17 @@ function RunBody({
>
Context
</TabButton>
<TabButton
isActive={tab === "metadata"}
layoutId="span-run"
onClick={() => {
replace({ tab: "metadata" });
}}
shortcut={{ key: "m" }}
>
Metadata
</TabButton>
</TabContainer>
</div>
<div className="overflow-y-auto px-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
@@ -607,6 +618,16 @@ function RunBody({
<div className="flex flex-col gap-4 py-3">
<CodeBlock code={run.context} showLineNumbers={false} />
</div>
) : tab === "metadata" ? (
<div className="flex flex-col gap-4 py-3">
{run.metadata ? (
<CodeBlock code={run.metadata} showLineNumbers={false} />
) : (
<Callout to="https://trigger.dev/docs/runs/metadata" variant="docs">
No metadata set for this run. View our metadata documentation to learn more.
</Callout>
)}
</div>
) : (
<div className="flex flex-col gap-4 pt-3">
<div className="border-b border-grid-bright pb-3">
apps/webapp/app/routes/resources.runs.$runParam.ts
+2
View File
@@ -72,6 +72,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
},
payload: true,
payloadType: true,
metadata: true,
metadataType: true,
maxAttempts: true,
project: {
include: {
apps/webapp/app/utils/packets.ts
+36
View File
@@ -0,0 +1,36 @@
import { IOPacket } from "@trigger.dev/core/v3/utils/ioSerialization";
import { env } from "~/env.server";
import { ServiceValidationError } from "~/v3/services/baseService.server";
export class MetadataTooLargeError extends ServiceValidationError {
constructor(message: string) {
super(message, 413);
this.name = "MetadataTooLargeError";
}
}
export function handleMetadataPacket(metadata: any, metadataType: string): IOPacket | undefined {
let metadataPacket: IOPacket | undefined = undefined;
if (typeof metadata === "string") {
metadataPacket = { data: metadata, dataType: metadataType };
}
if (metadataType === "application/json") {
metadataPacket = { data: JSON.stringify(metadata), dataType: "application/json" };
}
if (!metadataPacket || !metadataPacket.data) {
return;
}
const byteLength = Buffer.byteLength(metadataPacket.data, "utf8");
if (byteLength > env.TASK_RUN_METADATA_MAXIMUM_SIZE) {
throw new MetadataTooLargeError(
`Metadata exceeds maximum size of ${env.TASK_RUN_METADATA_MAXIMUM_SIZE} bytes`
);
}
return metadataPacket;
}
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
+7
View File
@@ -9,6 +9,7 @@ import {
TaskRunExecutionResult,
TaskRunFailedExecutionResult,
TaskRunSuccessfulExecutionResult,
parsePacket,
serverWebsocketMessages,
} from "@trigger.dev/core/v3";
import { ZodMessageSender } from "@trigger.dev/core/v3/zodMessageHandler";
@@ -1033,6 +1034,11 @@ class SharedQueueTasks {
const machinePreset = machinePresetFromConfig(backgroundWorkerTask.machineConfig ?? {});
const metadata = await parsePacket({
data: taskRun.metadata ?? undefined,
dataType: taskRun.metadataType,
});
const execution: ProdTaskRunExecution = {
task: {
id: backgroundWorkerTask.slug,
@@ -1060,6 +1066,7 @@ class SharedQueueTasks {
durationMs: taskRun.usageDurationMs,
costInCents: taskRun.costInCents,
baseCostInCents: taskRun.baseCostInCents,
metadata,
},
queue: {
id: queue.friendlyId,
apps/webapp/app/v3/services/completeAttempt.server.ts
+15
View File
@@ -128,6 +128,12 @@ export class CompleteAttemptService extends BaseService {
output: completion.output,
outputType: completion.outputType,
usageDurationMs: completion.usage?.durationMs,
taskRun: {
update: {
output: completion.output,
outputType: completion.outputType,
},
},
},
});
@@ -348,6 +354,15 @@ export class CompleteAttemptService extends BaseService {
})
);
} else {
await this._prisma.taskRun.update({
where: {
id: taskRunAttempt.taskRunId,
},
data: {
error: sanitizedError,
},
});
const finalizeService = new FinalizeTaskRunService();
await finalizeService.call({
id: taskRunAttempt.taskRunId,
apps/webapp/app/v3/services/createTaskRunAttempt.server.ts
+7 -1
View File
@@ -1,4 +1,4 @@
import { TaskRunExecution } from "@trigger.dev/core/v3";
import { parsePacket, TaskRunExecution } from "@trigger.dev/core/v3";
import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { logger } from "~/services/logger.server";
@@ -157,6 +157,11 @@ export class CreateTaskRunAttemptService extends BaseService {
const machinePreset = machinePresetFromConfig(taskRun.lockedBy.machineConfig ?? {});
const metadata = await parsePacket({
data: taskRun.metadata ?? undefined,
dataType: taskRun.metadataType,
});
const execution: TaskRunExecution = {
task: {
id: taskRun.lockedBy.slug,
@@ -186,6 +191,7 @@ export class CreateTaskRunAttemptService extends BaseService {
baseCostInCents: taskRun.baseCostInCents,
maxAttempts: taskRun.maxAttempts ?? undefined,
version: taskRun.lockedBy.worker.version,
metadata,
},
queue: {
id: queue.friendlyId,
apps/webapp/app/v3/services/finalizeTaskRun.server.ts
+19 -6
View File
@@ -2,15 +2,11 @@ import { sanitizeError, TaskRunError } from "@trigger.dev/core/v3";
import { type Prisma, type TaskRun } from "@trigger.dev/database";
import { logger } from "~/services/logger.server";
import { marqs, sanitizeQueueName } from "~/v3/marqs/index.server";
import {
isFailedRunStatus,
type FINAL_ATTEMPT_STATUSES,
type FINAL_RUN_STATUSES,
} from "../taskStatus";
import { generateFriendlyId } from "../friendlyIdentifiers";
import { FINAL_ATTEMPT_STATUSES, isFailedRunStatus, type FINAL_RUN_STATUSES } from "../taskStatus";
import { PerformTaskRunAlertsService } from "./alerts/performTaskRunAlerts.server";
import { BaseService } from "./baseService.server";
import { ResumeDependentParentsService } from "./resumeDependentParents.server";
import { generateFriendlyId } from "../friendlyIdentifiers";
type BaseInput = {
id: string;
@@ -70,6 +66,10 @@ export class FinalizeTaskRunService extends BaseService {
await this.finalizeAttempt({ attemptStatus, error, run });
}
if (isFailedRunStatus(run.status)) {
await this.finalizeRunError(run, error);
}
//resume any dependencies
const resumeService = new ResumeDependentParentsService(this._prisma);
const result = await resumeService.call({ id: run.id });
@@ -88,6 +88,19 @@ export class FinalizeTaskRunService extends BaseService {
return run as Output<T>;
}
async finalizeRunError(run: TaskRun, error?: TaskRunError) {
if (!error) {
return;
}
await this._prisma.taskRun.update({
where: { id: run.id },
data: {
error: sanitizeError(error),
},
});
}
async finalizeAttempt({
attemptStatus,
error,
apps/webapp/app/v3/services/replayTaskRun.server.ts
+8
View File
@@ -65,6 +65,13 @@ export class ReplayTaskRunService extends BaseService {
payloadPacketType: payloadPacket.dataType,
});
const metadata = existingTaskRun.metadata
? await parsePacket({
data: existingTaskRun.metadata,
dataType: existingTaskRun.metadataType,
})
: undefined;
try {
const tags = await getTagsForRunId({
friendlyId: existingTaskRun.friendlyId,
@@ -85,6 +92,7 @@ export class ReplayTaskRunService extends BaseService {
test: existingTaskRun.isTest,
payloadType: payloadPacket.dataType,
tags: tags?.map((t) => t.name) as RunTags,
metadata,
},
},
{
apps/webapp/app/v3/services/testTask.server.ts
+1
View File
@@ -19,6 +19,7 @@ export class TestTaskService extends BaseService {
payload: data.payload,
options: {
test: true,
metadata: data.metadata,
},
});
case "SCHEDULED": {
apps/webapp/app/v3/services/triggerTask.server.ts
+10
View File
@@ -20,6 +20,7 @@ import { logger } from "~/services/logger.server";
import { isFinalAttemptStatus, isFinalRunStatus } from "../taskStatus";
import { createTag, MAX_TAGS_PER_RUN } from "~/models/taskRunTag.server";
import { findCurrentWorkerFromEnvironment } from "../models/workerDeployment.server";
import { handleMetadataPacket } from "~/utils/packets";
export type TriggerTaskServiceOptions = {
idempotencyKey?: string;
@@ -99,6 +100,13 @@ export class TriggerTaskService extends BaseService {
environment
);
const metadataPacket = body.options?.metadata
? handleMetadataPacket(
body.options?.metadata,
body.options?.metadataType ?? "application/json"
)
: undefined;
const dependentAttempt = body.options?.dependentAttempt
? await this._prisma.taskRunAttempt.findUnique({
where: { friendlyId: body.options.dependentAttempt },
@@ -341,6 +349,8 @@ export class TriggerTaskService extends BaseService {
batchId: dependentBatchRun?.id ?? parentBatchRun?.id,
resumeParentOnCompletion: !!(dependentAttempt ?? dependentBatchRun),
depth,
metadata: metadataPacket?.data,
metadataType: metadataPacket?.dataType,
},
});
apps/webapp/app/v3/taskStatus.ts
+5
View File
@@ -50,8 +50,13 @@ export const FINAL_ATTEMPT_STATUSES = [
"COMPLETED",
"FAILED",
] satisfies TaskRunAttemptStatus[];
export type FINAL_ATTEMPT_STATUSES = (typeof FINAL_ATTEMPT_STATUSES)[number];
export const FAILED_ATTEMPT_STATUSES = ["FAILED", "CANCELED"] satisfies TaskRunAttemptStatus[];
export type FAILED_ATTEMPT_STATUSES = (typeof FAILED_ATTEMPT_STATUSES)[number];
export const FREEZABLE_RUN_STATUSES: TaskRunStatus[] = ["EXECUTING", "RETRYING_AFTER_FAILURE"];
export const FREEZABLE_ATTEMPT_STATUSES: TaskRunAttemptStatus[] = ["EXECUTING", "FAILED"];
apps/webapp/app/v3/testTask.ts
+20
View File
@@ -11,6 +11,26 @@ export const TestTaskData = z
} catch (e) {
console.log("parsing error", e);
if (e instanceof Error) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: e.message,
});
} else {
ctx.addIssue({
code: z.ZodIssueCode.custom,
message: "This is invalid JSON",
});
}
}
}),
metadata: z.string().transform((metadata, ctx) => {
try {
const data = JSON.parse(metadata);
return data as any;
} catch (e) {
console.log("parsing error", e);
if (e instanceof Error) {
ctx.addIssue({
code: z.ZodIssueCode.custom,
docs/images/run-metadata.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 666 KiB

docs/management/runs/update-metadata.mdx
+4
View File
@@ -0,0 +1,4 @@
---
title: "Update metadata"
openapi: "v3-openapi PUT /api/v1/runs/{runId}/metadata"
---
docs/mint.json
+54 -14
View File
@@ -1,7 +1,10 @@
{
"$schema": "https://mintlify.com/schema.json",
"name": "Trigger.dev",
"openapi": ["/openapi.yml", "/v3-openapi.yaml"],
"openapi": [
"/openapi.yml",
"/v3-openapi.yaml"
],
"api": {
"playground": {
"mode": "simple"
@@ -103,26 +106,41 @@
"navigation": [
{
"group": "Getting Started",
"pages": ["introduction", "quick-start", "how-it-works", "upgrading-beta", "limits"]
"pages": [
"introduction",
"quick-start",
"how-it-works",
"upgrading-beta",
"limits"
]
},
{
"group": "Fundamentals",
"pages": [
{
"group": "Tasks",
"pages": ["tasks/overview", "tasks/scheduled"]
"pages": [
"tasks/overview",
"tasks/scheduled"
]
},
"triggering",
"apikeys",
{
"group": "Configuration",
"pages": ["config/config-file", "config/extensions/overview"]
"pages": [
"config/config-file",
"config/extensions/overview"
]
}
]
},
{
"group": "Development",
"pages": ["cli-dev", "run-tests"]
"pages": [
"cli-dev",
"run-tests"
]
},
{
"group": "Deployment",
@@ -132,7 +150,9 @@
"github-actions",
{
"group": "Deployment integrations",
"pages": ["vercel-integration"]
"pages": [
"vercel-integration"
]
}
]
},
@@ -144,7 +164,13 @@
"errors-retrying",
{
"group": "Wait",
"pages": ["wait", "wait-for", "wait-until", "wait-for-event", "wait-for-request"]
"pages": [
"wait",
"wait-for",
"wait-until",
"wait-for-event",
"wait-for-request"
]
},
"queue-concurrency",
"versioning",
@@ -162,7 +188,10 @@
"management/overview",
{
"group": "Tasks API",
"pages": ["management/tasks/trigger", "management/tasks/batch-trigger"]
"pages": [
"management/tasks/trigger",
"management/tasks/batch-trigger"
]
},
{
"group": "Runs API",
@@ -171,7 +200,8 @@
"management/runs/retrieve",
"management/runs/replay",
"management/runs/cancel",
"management/runs/reschedule"
"management/runs/reschedule",
"management/runs/update-metadata"
]
},
{
@@ -200,7 +230,9 @@
},
{
"group": "Projects API",
"pages": ["management/projects/runs"]
"pages": [
"management/projects/runs"
]
}
]
},
@@ -246,7 +278,11 @@
},
{
"group": "Help",
"pages": ["community", "help-slack", "help-email"]
"pages": [
"community",
"help-slack",
"help-email"
]
},
{
"group": "Frameworks",
@@ -287,11 +323,15 @@
},
{
"group": "Dashboard",
"pages": ["guides/dashboard/creating-a-project"]
"pages": [
"guides/dashboard/creating-a-project"
]
},
{
"group": "Migrations",
"pages": ["guides/use-cases/upgrading-from-v2"]
"pages": [
"guides/use-cases/upgrading-from-v2"
]
}
],
"footerSocials": {
@@ -299,4 +339,4 @@
"github": "https://github.com/triggerdotdev",
"linkedin": "https://www.linkedin.com/company/triggerdotdev"
}
}
}
docs/runs/metadata.mdx
+273
View File
@@ -0,0 +1,273 @@
---
title: "Run metadata"
sidebarTitle: "Metadata"
description: "Attach a small amount of data to a run and update it as the run progresses."
---
You can attach up to 4KB (4,096 bytes) of metadata to a run, which you can then access from inside the run function, via the API, and in the dashboard. You can use metadata to store additional, structured information on a run. For example, you could store your user’s full name and corresponding unique identifier from your system on every task that is associated with that user.
## Usage
Add metadata to a run by passing it as an object to the `trigger` function:
```ts
const handle = await myTask.trigger(
{ message: "hello world" },
{ metadata: { user: { name: "Eric", id: "user_1234" } } }
);
```
Then inside your run function, you can access the metadata like this:
```ts
import { task, metadata } from "@trigger.dev/sdk/v3";
export const myTask = task({
id: "my-task",
run: async (payload: { message: string }) => {
const user = metadata.get("user");
console.log(user.name); // "Eric"
console.log(user.id); // "user_1234"
},
});
```
You can also update the metadata during the run:
```ts
import { task, metadata } from "@trigger.dev/sdk/v3";
export const myTask = task({
id: "my-task",
run: async (payload: { message: string }) => {
// Do some work
await metadata.set("progress", 0.1);
// Do some more work
await metadata.set("progress", 0.5);
// Do even more work
await metadata.set("progress", 1.0);
},
});
```
You can get the current metadata at any time by calling `metadata.get()` or `metadata.current()` (again, only inside a run):
```ts
import { task, metadata } from "@trigger.dev/sdk/v3";
export const myTask = task({
id: "my-task",
run: async (payload: { message: string }) => {
// Get the whole metadata object
const currentMetadata = metadata.current();
console.log(currentMetadata);
// Get a specific key
const user = metadata.get("user");
console.log(user.name); // "Eric"
},
});
```
You can update metadata inside a run using `metadata.set()`, `metadata.save()`, or `metadata.del()`:
```ts
import { task, metadata } from "@trigger.dev/sdk/v3";
export const myTask = task({
id: "my-task",
run: async (payload: { message: string }) => {
// Set a key
await metadata.set("progress", 0.5);
// Update the entire metadata object
await metadata.save({ progress: 0.6 });
// Delete a key
await metadata.del("progress");
},
});
```
Any of these methods can be called anywhere "inside" the run function, or a function called from the run function:
```ts
import { task, metadata } from "@trigger.dev/sdk/v3";
export const myTask = task({
id: "my-task",
run: async (payload: { message: string }) => {
await doSomeWork();
},
});
async function doSomeWork() {
await metadata.set("progress", 0.5);
}
```
If you call any of the metadata methods outside of the run function, they will have no effect:
```ts
import { metadata } from "@trigger.dev/sdk/v3";
// Somewhere outside of the run function
async function doSomeWork() {
await metadata.set("progress", 0.5); // This will do nothing
}
```
This means it's safe to call these methods anywhere in your code, and they will only have an effect when called inside the run function.
<Note>
Calling `metadata.current()` or `metadata.get()` outside of the run function will always return
undefined.
</Note>
These methods also work inside any task lifecycle hook, either attached to the specific task or the global hooks defined in your `trigger.config.ts` file.
<CodeGroup>
```ts myTasks.ts
import { task, metadata } from "@trigger.dev/sdk/v3";
export const myTask = task({
id: "my-task",
run: async (payload: { message: string }) => {
// Your run function work here
},
onStart: async () => {
await metadata.set("progress", 0.5);
},
onSuccess: async () => {
await metadata.set("progress", 1.0);
},
});
```
```ts trigger.config.ts
import { defineConfig, metadata } from "@trigger.dev/sdk/v3";
export default defineConfig({
project: "proj_1234",
onStart: async () => {
await metadata.set("progress", 0.5);
},
});
```
</CodeGroup>
## Metadata propagation
Metadata is NOT propagated to child tasks. If you want to pass metadata to a child task, you must do so explicitly:
```ts
import { task, metadata } from "@trigger.dev/sdk/v3";
export const myTask = task({
id: "my-task",
run: async (payload: { message: string }) => {
await metadata.set("progress", 0.5);
await childTask.trigger(payload, { metadata: metadata.current() });
},
});
```
## Type-safe metadata
The metadata APIs are currently loosely typed, accepting any object that is JSON-serializable:
```ts
// ❌ You can't pass a top-level array
const handle = await myTask.trigger(
{ message: "hello world" },
{ metadata: [{ user: { name: "Eric", id: "user_1234" } }] }
);
// ❌ You can't pass a string as the entire metadata:
const handle = await myTask.trigger(
{ message: "hello world" },
{ metadata: "this is the metadata" }
);
// ❌ You can't pass in a function or a class instance
const handle = await myTask.trigger(
{ message: "hello world" },
{ metadata: { user: () => "Eric", classInstance: new HelloWorld() } }
);
// ✅ You can pass in dates and other JSON-serializable objects
const handle = await myTask.trigger(
{ message: "hello world" },
{ metadata: { user: { name: "Eric", id: "user_1234" }, date: new Date() } }
);
```
<Note>
If you pass in an object like a Date, it will be serialized to a string when stored in the
metadata. That also means that when you retrieve it using `metadata.get()` or
`metadata.current()`, you will get a string back. You will need to deserialize it back to a Date
object if you need to use it as a Date.
</Note>
We recommend wrapping the metadata API in a [Zod](https://zod.dev) schema (or your validator library of choice) to provide type safety:
```ts
import { task, metadata } from "@trigger.dev/sdk/v3";
import { z } from "zod";
const Metadata = z.object({
user: z.object({
name: z.string(),
id: z.string(),
}),
date: z.coerce.date(), // Coerce the date string back to a Date object
});
type Metadata = z.infer<typeof Metadata>;
// Helper function to get the metadata object in a type-safe way
// Note: you would probably want to use .safeParse instead of .parse in a real-world scenario
function getMetadata() {
return Metadata.parse(metadata.current());
}
export const myTask = task({
id: "my-task",
run: async (payload: { message: string }) => {
const metadata = getMetadata();
console.log(metadata.user.name); // "Eric"
console.log(metadata.user.id); // "user_1234"
console.log(metadata.date); // Date object
},
});
```
## Inspecting metadata
### Dashboard
You can view the metadata for a run in the Trigger.dev dashboard. The metadata will be displayed in the run details view:
![View run metadata dashboard](/images/run-metadata.png)
### API
You can use the `runs.retrieve()` SDK function to get the metadata for a run:
```ts
import { runs } from "@trigger.dev/sdk/v3";
const run = await runs.retrieve("run_1234");
console.log(run.metadata);
```
See the [API reference](/management/runs/retrieve) for more information.
## Size limit
The maximum size of the metadata object is 4KB. If you exceed this limit, the SDK will throw an error. If you are self-hosting Trigger.dev, you can increase this limit by setting the `TASK_RUN_METADATA_MAXIMUM_SIZE` environment variable. For example, to increase the limit to 16KB, you would set `TASK_RUN_METADATA_MAXIMUM_SIZE=16384`.
docs/tasks/overview.mdx
+12 -9
View File
@@ -194,13 +194,14 @@ export const taskWithOnStart = task({
You can also define an `onStart` function in your `trigger.config.ts` file to get notified when any task starts.
```ts trigger.config.ts
import type { TriggerConfig } from "@trigger.dev/sdk/v3";
import { defineConfig } from "@trigger.dev/sdk/v3";
export const config: TriggerConfig = {
export default defineConfig({
project: "proj_1234",
onStart: async (payload, { ctx }) => {
console.log("Task started", ctx.task.id);
},
};
});
```
### `onSuccess` function
@@ -222,13 +223,14 @@ export const taskWithOnSuccess = task({
You can also define an `onSuccess` function in your `trigger.config.ts` file to get notified when any task succeeds.
```ts trigger.config.ts
import type { TriggerConfig } from "@trigger.dev/sdk/v3";
import { defineConfig } from "@trigger.dev/sdk/v3";
export const config: TriggerConfig = {
export default defineConfig({
project: "proj_1234",
onSuccess: async (payload, output, { ctx }) => {
console.log("Task succeeded", ctx.task.id);
},
};
});
```
### `onFailure` function
@@ -250,13 +252,14 @@ export const taskWithOnFailure = task({
You can also define an `onFailure` function in your `trigger.config.ts` file to get notified when any task fails.
```ts trigger.config.ts
import type { TriggerConfig } from "@trigger.dev/sdk/v3";
import { defineConfig } from "@trigger.dev/sdk/v3";
export const config: TriggerConfig = {
export default defineConfig({
project: "proj_1234",
onFailure: async (payload, error, { ctx }) => {
console.log("Task failed", ctx.task.id);
},
};
});
```
### `handleError` functions
docs/triggering.mdx
+9
View File
@@ -220,6 +220,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
return json(result);
}
```
</CodeGroup>
<Note>
@@ -733,6 +734,14 @@ await myTask.trigger({ some: "data" }, { maxAttempts: 1 }); // no retries
This will override the `retry.maxAttempts` value set in the task definition.
### `tags`
View our [tags doc](/tags) for more information.
### `metadata`
View our [metadata doc](/runs/metadata) for more information.
## Large Payloads
We recommend keeping your task payloads as small as possible. We currently have a hard limit on task payloads above 10MB.
docs/v3-openapi.yaml
+84
View File
@@ -360,6 +360,86 @@ paths:
const handle = await runs.replay("run_1234");
"/api/v1/runs/{runId}/metadata":
parameters:
- $ref: "#/components/parameters/runId"
put:
operationId: update_run_metadata_v1
summary: Update run metadata
description: Update the metadata of a run.
requestBody:
required: true
content:
application/json:
schema:
type: object
properties:
metadata:
type: object
description: The new metadata to set on the run.
example: { key: "value" }
responses:
"200":
description: Successful request
content:
application/json:
schema:
type: object
properties:
metadata:
type: object
description: The updated metadata of the run.
"400":
description: Invalid request
content:
application/json:
schema:
type: object
properties:
error:
type: string
enum:
- Invalid or missing run ID
- Invalid metadata
"401":
description: Unauthorized request
content:
application/json:
schema:
type: object
properties:
error:
type: string
enum:
- Invalid or Missing API key
"404":
description: Resource not found
content:
application/json:
schema:
type: object
properties:
error:
type: string
enum:
- Task Run not found
tags:
- runs
security:
- secretKey: []
x-codeSamples:
- lang: typescript
label: Update metadata
source: |-
import { metadata, task } from "@trigger.dev/sdk/v3";
export const myTask = task({
id: "my-task",
run: async () => {
await metadata.update({ key: "value" });
}
});
"/api/v2/runs/{runId}/cancel":
parameters:
- $ref: "#/components/parameters/runId"
@@ -1897,6 +1977,10 @@ components:
items:
type: string
description: A tag must be between 1 and 64 characters, a run can have up to 5 tags attached to it.
metadata:
type: object
description: The metadata of the run. See [Metadata](/runs/metadata) for more information.
example: { "foo": "bar" }
costInCents:
type: number
example: 0.00292
packages/core/src/v3/apiClient/index.ts
+19
View File
@@ -22,6 +22,8 @@ import {
TriggerTaskRequestBody,
TriggerTaskResponse,
UpdateEnvironmentVariableRequestBody,
UpdateMetadataRequestBody,
UpdateMetadataResponseBody,
UpdateScheduleOptions,
} from "../schemas/index.js";
import { taskContext } from "../task-context-api.js";
@@ -498,6 +500,23 @@ export class ApiClient {
);
}
updateRunMetadata(
runId: string,
body: UpdateMetadataRequestBody,
requestOptions?: ZodFetchOptions
) {
return zodfetch(
UpdateMetadataResponseBody,
`${this.baseUrl}/api/v1/runs/${runId}/metadata`,
{
method: "PUT",
headers: this.#getHeaders(false),
body: JSON.stringify(body),
},
mergeRequestOptions(this.defaultRequestOptions, requestOptions)
);
}
#getHeaders(spanParentAsLink: boolean) {
const headers: Record<string, string> = {
"Content-Type": "application/json",
packages/core/src/v3/apiClientManager/index.ts
+29
View File
@@ -5,6 +5,13 @@ import { ApiClientConfiguration } from "./types.js";
const API_NAME = "api-client";
export class ApiClientMissingError extends Error {
constructor(message: string) {
super(message);
this.name = "ApiClientMissingError";
}
}
export class APIClientManagerAPI {
private static _instance?: APIClientManagerAPI;
@@ -44,7 +51,29 @@ export class APIClientManagerAPI {
return new ApiClient(this.baseURL, this.accessToken);
}
clientOrThrow(): ApiClient {
if (!this.baseURL || !this.accessToken) {
throw new ApiClientMissingError(this.apiClientMissingError());
}
return new ApiClient(this.baseURL, this.accessToken);
}
#getConfig(): ApiClientConfiguration | undefined {
return getGlobal(API_NAME);
}
apiClientMissingError() {
const hasBaseUrl = !!this.baseURL;
const hasAccessToken = !!this.accessToken;
if (!hasBaseUrl && !hasAccessToken) {
return `You need to set the TRIGGER_API_URL and TRIGGER_SECRET_KEY environment variables. See https://trigger.dev/docs/management/overview#authentication`;
} else if (!hasBaseUrl) {
return `You need to set the TRIGGER_API_URL environment variable. See https://trigger.dev/docs/management/overview#authentication`;
} else if (!hasAccessToken) {
return `You need to set the TRIGGER_SECRET_KEY environment variable. See https://trigger.dev/docs/management/overview#authentication`;
}
return `Unknown error`;
}
}
packages/core/src/v3/index.ts
+1
View File
@@ -11,6 +11,7 @@ export * from "./runtime-api.js";
export * from "./task-context-api.js";
export * from "./apiClientManager-api.js";
export * from "./usage-api.js";
export * from "./run-metadata-api.js";
export * from "./schemas/index.js";
export { SemanticInternalAttributes } from "./semanticInternalAttributes.js";
export * from "./task-catalog-api.js";
packages/core/src/v3/run-metadata-api.ts
+5
View File
@@ -0,0 +1,5 @@
// Split module-level variable definition into separate files to allow
// tree-shaking on each api instance.
import { RunMetadataAPI } from "./runMetadata/index.js";
export const runMetadata = RunMetadataAPI.getInstance();
packages/core/src/v3/runMetadata/index.ts
+96
View File
@@ -0,0 +1,96 @@
import { DeserializedJson } from "../../schemas/json.js";
import { apiClientManager } from "../apiClientManager-api.js";
import { taskContext } from "../task-context-api.js";
import { ApiRequestOptions } from "../zodfetch.js";
export class RunMetadataAPI {
private static _instance?: RunMetadataAPI;
private store: Record<string, DeserializedJson> | undefined;
private constructor() {}
public static getInstance(): RunMetadataAPI {
if (!this._instance) {
this._instance = new RunMetadataAPI();
}
return this._instance;
}
public enterWithMetadata(metadata: Record<string, DeserializedJson>): void {
this.store = metadata;
}
public current(): Record<string, DeserializedJson> | undefined {
return this.store;
}
public getKey(key: string): DeserializedJson | undefined {
return this.store?.[key];
}
public async setKey(
key: string,
value: DeserializedJson,
requestOptions?: ApiRequestOptions
): Promise<void> {
const runId = taskContext.ctx?.run.id;
if (!runId) {
return;
}
const apiClient = apiClientManager.clientOrThrow();
const nextStore = {
...(this.store ?? {}),
[key]: value,
};
const response = await apiClient.updateRunMetadata(
runId,
{ metadata: nextStore },
requestOptions
);
this.store = response.metadata;
}
public async deleteKey(key: string, requestOptions?: ApiRequestOptions): Promise<void> {
const runId = taskContext.ctx?.run.id;
if (!runId) {
return;
}
const apiClient = apiClientManager.clientOrThrow();
const nextStore = { ...(this.store ?? {}) };
delete nextStore[key];
const response = await apiClient.updateRunMetadata(
runId,
{ metadata: nextStore },
requestOptions
);
this.store = response.metadata;
}
public async update(
metadata: Record<string, DeserializedJson>,
requestOptions?: ApiRequestOptions
): Promise<void> {
const runId = taskContext.ctx?.run.id;
if (!runId) {
return;
}
const apiClient = apiClientManager.clientOrThrow();
const response = await apiClient.updateRunMetadata(runId, { metadata }, requestOptions);
this.store = response.metadata;
}
}
packages/core/src/v3/schemas/api.ts
+17
View File
@@ -2,6 +2,7 @@ import { z } from "zod";
import { BackgroundWorkerMetadata } from "./resources.js";
import { QueueOptions } from "./schemas.js";
import { SerializedError } from "./common.js";
import { DeserializedJsonSchema, SerializableJsonSchema } from "../../schemas/json.js";
export const WhoAmIResponseSchema = z.object({
userId: z.string(),
@@ -81,6 +82,8 @@ export const TriggerTaskRequestBody = z.object({
ttl: z.string().or(z.number().nonnegative().int()).optional(),
tags: RunTags.optional(),
maxAttempts: z.number().int().optional(),
metadata: z.any(),
metadataType: z.string().optional(),
})
.optional(),
});
@@ -505,6 +508,7 @@ const CommonRunFields = {
costInCents: z.number(),
baseCostInCents: z.number(),
durationMs: z.number(),
metadata: z.record(z.any()).optional(),
};
const RetrieveRunCommandFields = {
@@ -608,3 +612,16 @@ export const EnvironmentVariable = z.object({
export const EnvironmentVariables = z.array(EnvironmentVariable);
export type EnvironmentVariables = z.infer<typeof EnvironmentVariables>;
export const UpdateMetadataRequestBody = z.object({
metadata: z.record(DeserializedJsonSchema),
metadataType: z.string().optional(),
});
export type UpdateMetadataRequestBody = z.infer<typeof UpdateMetadataRequestBody>;
export const UpdateMetadataResponseBody = z.object({
metadata: z.record(DeserializedJsonSchema),
});
export type UpdateMetadataResponseBody = z.infer<typeof UpdateMetadataResponseBody>;
packages/core/src/v3/schemas/common.ts
+3 -1
View File
@@ -1,4 +1,5 @@
import { z } from "zod";
import { DeserializedJsonSchema } from "../../schemas/json.js";
// Defaults to 0.5
export const MachineCpu = z.union([
@@ -142,6 +143,7 @@ export const TaskRun = z.object({
costInCents: z.number().default(0),
baseCostInCents: z.number().default(0),
version: z.string().optional(),
metadata: z.record(DeserializedJsonSchema).optional(),
});
export type TaskRun = z.infer<typeof TaskRun>;
@@ -221,7 +223,7 @@ export const TaskRunContext = z.object({
backgroundWorkerId: true,
backgroundWorkerTaskId: true,
}),
run: TaskRun.omit({ payload: true, payloadType: true }),
run: TaskRun.omit({ payload: true, payloadType: true, metadata: true }),
queue: TaskRunExecutionQueue,
environment: TaskRunExecutionEnvironment,
organization: TaskRunExecutionOrganization,
packages/core/src/v3/workers/taskExecutor.ts
+5 -2
View File
@@ -3,7 +3,7 @@ import { VERSION } from "../../version.js";
import { ApiError, RateLimitError } from "../apiClient/errors.js";
import { ConsoleInterceptor } from "../consoleInterceptor.js";
import { parseError, sanitizeError } from "../errors.js";
import { TriggerConfig } from "../index.js";
import { runMetadata, TriggerConfig } from "../index.js";
import { recordSpanException, TracingSDK } from "../otel/index.js";
import {
ServerBackgroundWorker,
@@ -26,7 +26,6 @@ import {
stringifyIO,
} from "../utils/ioSerialization.js";
import { calculateNextRetryDelay } from "../utils/retries.js";
import { accessoryAttributes } from "../utils/styleAttributes.js";
export type TaskExecutorOptions = {
tracingSDK: TracingSDK;
@@ -73,6 +72,10 @@ export class TaskExecutor {
worker,
});
if (execution.run.metadata) {
runMetadata.enterWithMetadata(execution.run.metadata);
}
this._tracingSDK.asyncResourceDetector.resolveWithAttributes({
...taskContext.attributes,
[SemanticInternalAttributes.SDK_VERSION]: VERSION,
packages/database/prisma/migrations/20240925092304_add_metadata_and_output_to_task_run/migration.sql
+5
View File
@@ -0,0 +1,5 @@
-- AlterTable
ALTER TABLE "TaskRun" ADD COLUMN "metadata" TEXT,
ADD COLUMN "metadataType" TEXT NOT NULL DEFAULT 'application/json',
ADD COLUMN "output" TEXT,
ADD COLUMN "outputType" TEXT NOT NULL DEFAULT 'application/json';
packages/database/prisma/migrations/20240925205409_add_error_to_task_run/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "TaskRun" ADD COLUMN "error" JSONB;
packages/database/prisma/schema.prisma
+11
View File
@@ -1751,6 +1751,17 @@ model TaskRun {
/// The span ID of the "trigger" span in the parent task run
parentSpanId String?
/// Run metadata
metadata String?
metadataType String @default("application/json")
/// Run output
output String?
outputType String @default("application/json")
/// Run error
error Json?
@@unique([runtimeEnvironmentId, taskIdentifier, idempotencyKey])
// Finding child runs
@@index([parentTaskRunId])
packages/trigger-sdk/src/v3/envvars.ts
+6 -31
View File
@@ -14,7 +14,6 @@ import {
mergeRequestOptions,
taskContext,
} from "@trigger.dev/core/v3";
import { apiClientMissingError } from "./shared.js";
import { tracer } from "./tracer.js";
export type { CreateEnvironmentVariableParams, ImportEnvironmentVariablesParams };
@@ -76,11 +75,7 @@ export function upload(
$params = params;
}
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
return apiClient.importEnvVars($projectRef, $slug, $params, $requestOptions);
}
@@ -121,11 +116,7 @@ export function list(
$requestOptions
);
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
return apiClient.listEnvVars($projectRef, $slug, $requestOptions);
}
@@ -187,11 +178,7 @@ export function create(
$params = params;
}
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
return apiClient.createEnvVar($projectRef, $slug, $params, $requestOptions);
}
@@ -238,11 +225,7 @@ export function retrieve(
throw new Error("slug is required");
}
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
return apiClient.retrieveEnvVar($projectRef, $slug, $name, $requestOptions);
}
@@ -289,11 +272,7 @@ export function del(
throw new Error("slug is required");
}
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
return apiClient.deleteEnvVar($projectRef, $slug, $name, $requestOptions);
}
@@ -362,11 +341,7 @@ export function update(
$params = params;
}
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
return apiClient.updateEnvVar($projectRef, $slug, $name, $params, $requestOptions);
}
packages/trigger-sdk/src/v3/index.ts
+1
View File
@@ -7,6 +7,7 @@ export * from "./wait.js";
export * from "./usage.js";
export * from "./idempotencyKeys.js";
export * from "./tags.js";
export * from "./metadata.js";
export type { Context };
import type { Context } from "./shared.js";
packages/trigger-sdk/src/v3/metadata.ts
+164
View File
@@ -0,0 +1,164 @@
import { DeserializedJson } from "@trigger.dev/core";
import {
accessoryAttributes,
ApiRequestOptions,
flattenAttributes,
mergeRequestOptions,
runMetadata,
} from "@trigger.dev/core/v3";
import { tracer } from "./tracer.js";
/**
* Provides access to run metadata operations.
* @namespace
* @property {Function} current - Get the current run's metadata.
* @property {Function} get - Get a specific key from the current run's metadata.
* @property {Function} set - Set a key in the current run's metadata.
* @property {Function} del - Delete a key from the current run's metadata.
* @property {Function} save - Update the entire metadata object for the current run.
*/
export const metadata = {
current: currentMetadata,
get: getMetadataKey,
set: setMetadataKey,
del: deleteMetadataKey,
save: saveMetadata,
};
export type RunMetadata = Record<string, DeserializedJson>;
/**
* Returns the metadata of the current run if inside a task run.
* This function allows you to access the entire metadata object for the current run.
*
* @returns {RunMetadata | undefined} The current run's metadata or undefined if not in a run context.
*
* @example
* const currentMetadata = metadata.current();
* console.log(currentMetadata);
*/
function currentMetadata(): RunMetadata | undefined {
return runMetadata.current();
}
/**
* Get a specific key from the metadata of the current run if inside a task run.
*
* @param {string} key - The key to retrieve from the metadata.
* @returns {DeserializedJson | undefined} The value associated with the key, or undefined if not found or not in a run context.
*
* @example
* const user = metadata.get("user");
* console.log(user.name); // "Eric"
* console.log(user.id); // "user_1234"
*/
function getMetadataKey(key: string): DeserializedJson | undefined {
return runMetadata.getKey(key);
}
/**
* Set a key in the metadata of the current run if inside a task run.
* This function allows you to update or add a new key-value pair to the run's metadata.
*
* @param {string} key - The key to set in the metadata.
* @param {DeserializedJson} value - The value to associate with the key.
* @param {ApiRequestOptions} [requestOptions] - Optional API request options.
* @returns {Promise<void>} A promise that resolves when the metadata is updated.
*
* @example
* await metadata.set("progress", 0.5);
*/
async function setMetadataKey(
key: string,
value: DeserializedJson,
requestOptions?: ApiRequestOptions
): Promise<void> {
const $requestOptions = mergeRequestOptions(
{
tracer,
name: "metadata.set()",
icon: "code-plus",
attributes: {
...accessoryAttributes({
items: [
{
text: key,
variant: "normal",
},
],
style: "codepath",
}),
...flattenAttributes(value, key),
},
},
requestOptions
);
await runMetadata.setKey(key, value, $requestOptions);
}
/**
* Delete a key from the metadata of the current run if inside a task run.
*
* @param {string} key - The key to delete from the metadata.
* @param {ApiRequestOptions} [requestOptions] - Optional API request options.
* @returns {Promise<void>} A promise that resolves when the key is deleted from the metadata.
*
* @example
* await metadata.del("progress");
*/
async function deleteMetadataKey(key: string, requestOptions?: ApiRequestOptions): Promise<void> {
const $requestOptions = mergeRequestOptions(
{
tracer,
name: "metadata.del()",
icon: "code-minus",
attributes: {
...accessoryAttributes({
items: [
{
text: key,
variant: "normal",
},
],
style: "codepath",
}),
key,
},
},
requestOptions
);
await runMetadata.deleteKey(key, $requestOptions);
}
/**
* Update the entire metadata object for the current run if inside a task run.
* This function allows you to replace the entire metadata object with a new one.
*
* @param {RunMetadata} metadata - The new metadata object to set for the run.
* @param {ApiRequestOptions} [requestOptions] - Optional API request options.
* @returns {Promise<void>} A promise that resolves when the metadata is updated.
*
* @example
* await metadata.save({ progress: 0.6, user: { name: "Alice", id: "user_5678" } });
*/
async function saveMetadata(
metadata: RunMetadata,
requestOptions?: ApiRequestOptions
): Promise<void> {
const $requestOptions = mergeRequestOptions(
{
tracer,
name: "metadata.save()",
icon: "code-plus",
attributes: {
...flattenAttributes(metadata),
},
},
requestOptions
);
await runMetadata.update(metadata, $requestOptions);
}
packages/trigger-sdk/src/v3/runs.ts
+6 -26
View File
@@ -18,7 +18,7 @@ import {
isRequestOptions,
mergeRequestOptions,
} from "@trigger.dev/core/v3";
import { AnyTask, Prettify, RunHandle, Task, apiClientMissingError } from "./shared.js";
import { AnyTask, Prettify, RunHandle, Task } from "./shared.js";
import { tracer } from "./tracer.js";
import { resolvePresignedPacketUrl } from "@trigger.dev/core/v3/utils/ioSerialization";
@@ -57,11 +57,7 @@ function listRuns(
paramsOrOptions?: ListRunsQueryParams | ListProjectRunsQueryParams | ApiRequestOptions,
requestOptions?: ApiRequestOptions
): CursorPagePromise<typeof ListRunResponseItem> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = listRunsRequestOptions(
paramsOrProjectRef,
@@ -158,11 +154,7 @@ function retrieveRun<TRunId extends RunHandle<any> | AnyTask | string>(
runId: RunId<TRunId>,
requestOptions?: ApiRequestOptions
): ApiPromise<RetrieveRunResult<TRunId>> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
@@ -216,11 +208,7 @@ function replayRun(
runId: string,
requestOptions?: ApiRequestOptions
): ApiPromise<ReplayRunResponse> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
@@ -250,11 +238,7 @@ function cancelRun(
runId: string,
requestOptions?: ApiRequestOptions
): ApiPromise<CanceledRunResponse> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
@@ -285,11 +269,7 @@ function rescheduleRun(
body: RescheduleRunRequestBody,
requestOptions?: ApiRequestOptions
): ApiPromise<RetrieveRunResponse> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
packages/trigger-sdk/src/v3/schedules/index.ts
+8 -40
View File
@@ -12,7 +12,7 @@ import {
taskCatalog,
} from "@trigger.dev/core/v3";
import { zodfetch } from "@trigger.dev/core/v3/zodfetch";
import { Task, TaskOptions, apiClientMissingError, createTask } from "../shared.js";
import { Task, TaskOptions, createTask } from "../shared.js";
import * as SchedulesAPI from "./api.js";
import { tracer } from "../tracer.js";
@@ -86,11 +86,7 @@ export function create(
options: SchedulesAPI.CreateScheduleOptions,
requestOptions?: ApiRequestOptions
): ApiPromise<ScheduleObject> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
@@ -124,11 +120,7 @@ export function retrieve(
scheduleId: string,
requestOptions?: ApiRequestOptions
): ApiPromise<ScheduleObject> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
@@ -169,11 +161,7 @@ export function update(
options: SchedulesAPI.UpdateScheduleOptions,
requestOptions?: ApiRequestOptions
): ApiPromise<ScheduleObject> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
@@ -207,11 +195,7 @@ export function del(
scheduleId: string,
requestOptions?: ApiRequestOptions
): ApiPromise<DeletedScheduleObject> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
@@ -245,11 +229,7 @@ export function deactivate(
scheduleId: string,
requestOptions?: ApiRequestOptions
): ApiPromise<ScheduleObject> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
@@ -283,11 +263,7 @@ export function activate(
scheduleId: string,
requestOptions?: ApiRequestOptions
): ApiPromise<ScheduleObject> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
@@ -324,11 +300,7 @@ export function list(
options?: SchedulesAPI.ListScheduleOptions,
requestOptions?: ApiRequestOptions
): OffsetLimitPagePromise<typeof ScheduleObject> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const $requestOptions = mergeRequestOptions(
{
@@ -349,10 +321,6 @@ export function list(
export function timezones(options?: { excludeUtc?: boolean }) {
const baseUrl = apiClientManager.baseURL;
if (!baseUrl) {
throw apiClientMissingError();
}
return zodfetch(
TimezonesResult,
`${baseUrl}/api/v1/timezones${options?.excludeUtc === true ? "?excludeUtc=true" : ""}`,
packages/trigger-sdk/src/v3/shared.ts
+14 -34
View File
@@ -39,6 +39,7 @@ import {
import { IdempotencyKey, idempotencyKeys, isIdempotencyKey } from "./idempotencyKeys.js";
import { PollOptions, RetrieveRunResult, runs } from "./runs.js";
import { tracer } from "./tracer.js";
import { SerializableJson } from "@trigger.dev/core";
export type Context = TaskRunContext;
@@ -496,6 +497,11 @@ export type TaskRunOptions = {
* ```
*/
tags?: RunTags;
/**
* Metadata to attach to the run. Metadata can be used to store additional information about the run. Limited to 4KB.
*/
metadata?: Record<string, SerializableJson>;
};
type TaskRunConcurrencyOptions = Queue;
@@ -769,11 +775,7 @@ async function trigger_internal<TPayload, TOutput>(
options?: TaskRunOptions,
requestOptions?: ApiRequestOptions
): Promise<RunHandle<TOutput>> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const payloadPacket = await stringifyIO(payload);
@@ -792,6 +794,7 @@ async function trigger_internal<TPayload, TOutput>(
tags: options?.tags,
maxAttempts: options?.maxAttempts,
parentAttempt: taskContext.ctx?.attempt.id,
metadata: options?.metadata,
},
},
{
@@ -837,11 +840,7 @@ async function batchTrigger_internal<TPayload, TOutput>(
requestOptions?: ApiRequestOptions,
queue?: QueueOptions
): Promise<BatchRunHandle<TOutput>> {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const response = await apiClient.batchTriggerTask(
id,
@@ -863,6 +862,7 @@ async function batchTrigger_internal<TPayload, TOutput>(
tags: item.options?.tags,
maxAttempts: item.options?.maxAttempts,
parentAttempt: taskContext.ctx?.attempt.id,
metadata: item.options?.metadata,
},
};
})
@@ -912,11 +912,7 @@ async function triggerAndWait_internal<TPayload, TOutput>(
throw new Error("triggerAndWait can only be used from inside a task.run()");
}
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const payloadPacket = await stringifyIO(payload);
@@ -939,6 +935,7 @@ async function triggerAndWait_internal<TPayload, TOutput>(
ttl: options?.ttl,
tags: options?.tags,
maxAttempts: options?.maxAttempts,
metadata: options?.metadata,
},
},
{},
@@ -1006,11 +1003,7 @@ async function batchTriggerAndWait_internal<TPayload, TOutput>(
throw new Error("batchTriggerAndWait can only be used from inside a task.run()");
}
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
return await tracer.startActiveSpan(
name,
@@ -1035,6 +1028,7 @@ async function batchTriggerAndWait_internal<TPayload, TOutput>(
ttl: item.options?.ttl,
tags: item.options?.tags,
maxAttempts: item.options?.maxAttempts,
metadata: item.options?.metadata,
},
};
})
@@ -1194,20 +1188,6 @@ async function handleTaskRunExecutionResult<TOutput>(
}
}
export function apiClientMissingError() {
const hasBaseUrl = !!apiClientManager.baseURL;
const hasAccessToken = !!apiClientManager.accessToken;
if (!hasBaseUrl && !hasAccessToken) {
return `You need to set the TRIGGER_API_URL and TRIGGER_SECRET_KEY environment variables.`;
} else if (!hasBaseUrl) {
return `You need to set the TRIGGER_API_URL environment variable.`;
} else if (!hasAccessToken) {
return `You need to set the TRIGGER_SECRET_KEY environment variable.`;
}
return `Unknown error`;
}
async function makeKey(
idempotencyKey?: IdempotencyKey | string | string[]
): Promise<IdempotencyKey | undefined> {
packages/trigger-sdk/src/v3/tags.ts
+1 -6
View File
@@ -7,7 +7,6 @@ import {
mergeRequestOptions,
taskContext,
} from "@trigger.dev/core/v3";
import { apiClientMissingError } from "./shared.js";
import { tracer } from "./tracer.js";
export const tags = {
@@ -15,11 +14,7 @@ export const tags = {
};
async function addTags(tags: RunTags, requestOptions?: ApiRequestOptions) {
const apiClient = apiClientManager.client;
if (!apiClient) {
throw apiClientMissingError();
}
const apiClient = apiClientManager.clientOrThrow();
const run = taskContext.ctx?.run;
if (!run) {
references/v3-catalog/src/trigger/runMetadata.ts
+66
View File
@@ -0,0 +1,66 @@
import { logger, task, metadata, AbortTaskRunError } from "@trigger.dev/sdk/v3";
export const runMetadataTask = task({
id: "run-metadata-task",
run: async (payload: any) => {
await runMetadataChildTask.triggerAndWait(payload, {
metadata: {
hello: "world",
date: new Date(),
anotherThing: {
a: 1,
b: 2,
},
},
});
},
});
export const runMetadataChildTask = task({
id: "run-metadata-child-task",
run: async (payload: any, { ctx }) => {
await metadata.set("child", "task");
logger.info("metadata", { metadata: metadata.current() });
await metadata.set("child-2", "task-2");
logger.info("metadata", { current: metadata.current() });
await metadata.del("hello");
logger.info("metadata", { metadata: metadata.current() });
await metadata.save({
there: {
is: {
something: "here",
},
},
});
await runMetadataChildTask2.triggerAndWait(payload, {
metadata: metadata.current(),
});
return metadata.current();
},
onStart: async () => {
logger.info("metadata", { metadata: metadata.current() });
},
onSuccess: async () => {
logger.info("metadata", { metadata: metadata.current() });
},
});
export const runMetadataChildTask2 = task({
id: "run-metadata-child-task-2",
run: async (payload: any, { ctx }) => {
throw new AbortTaskRunError("aborting");
},
});
export const myTask = task({
id: "my-task",
run: async (payload: any) => {},
});