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

feat: run.ctx tidying and additions (#2322)

Browse Source 
* Cleanup context and execution creation, cache stuff, add parent and root task run ids

* more efficient by using friendly IDs instead of doing joins

* metadata.root/parent now reference current run when run has no root/parent

* Adding changeset

* try to make test less flaky

* Clean imports

* Another attempt to fix the flaky test

* Fix usage by still passing durationMs and costInCents to the execution, just not the run.ctx
This commit is contained in:
Eric Allam
2025-07-30 13:47:12 +01:00
committed by GitHub
parent 5ea6605ec6
commit 14dcc76f93
38 changed files with 1478 additions and 433 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/giant-plums-smash.md
+18
View File
@@ -0,0 +1,18 @@
---
"@trigger.dev/sdk": patch
---
Added and cleaned up the run ctx param:
- New optional properties `ctx.run.parentTaskRunId` and `ctx.run.rootTaskRunId` reference the current run's root/parent ID.
- Removed deprecated properties from `ctx`
- Added a new `ctx.deployment` object that contains information about the deployment associated with the run.
We also update `metadata.root` and `metadata.parent` to work even when the run is a "root" run (meaning it doesn't have a parent or a root associated run). This now works:
```ts
metadata.root.set("foo", "bar");
metadata.parent.set("baz", 1);
metadata.current().foo // "bar"
metadata.current().baz // 1
```
apps/webapp/app/presenters/v3/SpanPresenter.server.ts
+245 -184
View File
@@ -1,22 +1,29 @@
import {
type MachinePresetName,
MachinePreset,
prettyPrintPacket,
SemanticInternalAttributes,
TaskRunContext,
TaskRunError,
V3TaskRunContext,
} from "@trigger.dev/core/v3";
import { getMaxDuration } from "@trigger.dev/core/v3/isomorphic";
import { AttemptId, getMaxDuration } from "@trigger.dev/core/v3/isomorphic";
import { RUNNING_STATUSES } from "~/components/runs/v3/TaskRunStatus";
import { logger } from "~/services/logger.server";
import { eventRepository, rehydrateAttribute } from "~/v3/eventRepository.server";
import { machinePresetFromName, machinePresetFromRun } from "~/v3/machinePresets.server";
import { machinePresetFromRun } from "~/v3/machinePresets.server";
import { getTaskEventStoreTableForRun, type TaskEventStoreTable } from "~/v3/taskEventStore.server";
import { isFailedRunStatus, isFinalRunStatus } from "~/v3/taskStatus";
import { BasePresenter } from "./basePresenter.server";
import { WaitpointPresenter } from "./WaitpointPresenter.server";
import { engine } from "~/v3/runEngine.server";
type Result = Awaited<ReturnType<SpanPresenter["call"]>>;
export type Span = NonNullable<NonNullable<Result>["span"]>;
export type SpanRun = NonNullable<NonNullable<Result>["run"]>;
type FindRunResult = NonNullable<
Awaited<ReturnType<InstanceType<typeof SpanPresenter>["findRun"]>>
>;
type GetSpanResult = NonNullable<Awaited<ReturnType<(typeof eventRepository)["getSpan"]>>>;
export class SpanPresenter extends BasePresenter {
public async call({
@@ -60,7 +67,7 @@ export class SpanPresenter extends BasePresenter {
const eventStore = getTaskEventStoreTableForRun(parentRun);
const run = await this.#getRun({
const run = await this.getRun({
eventStore,
traceId,
spanId,
@@ -95,7 +102,7 @@ export class SpanPresenter extends BasePresenter {
};
}
async #getRun({
async getRun({
eventStore,
traceId,
spanId,
@@ -120,6 +127,146 @@ export class SpanPresenter extends BasePresenter {
return;
}
const run = await this.findRun({ span, spanId });
if (!run) {
return;
}
const isFinished = isFinalRunStatus(run.status);
const output = !isFinished
? undefined
: run.outputType === "application/store"
? `/resources/packets/${run.runtimeEnvironment.id}/${run.output}`
: typeof run.output !== "undefined" && run.output !== null
? await prettyPrintPacket(run.output, run.outputType ?? undefined)
: undefined;
const payload =
run.payloadType === "application/store"
? `/resources/packets/${run.runtimeEnvironment.id}/${run.payload}`
: typeof run.payload !== "undefined" && run.payload !== null
? await prettyPrintPacket(run.payload, run.payloadType ?? undefined)
: undefined;
let error: TaskRunError | undefined = undefined;
if (run?.error) {
const result = TaskRunError.safeParse(run.error);
if (result.success) {
error = result.data;
} else {
error = {
type: "CUSTOM_ERROR",
raw: JSON.stringify(run.error),
};
}
}
const metadata = run.metadata
? await prettyPrintPacket(run.metadata, run.metadataType, {
filteredKeys: ["$$streams", "$$streamsVersion", "$$streamsBaseUrl"],
})
: undefined;
const machine = run.machinePreset ? machinePresetFromRun(run) : undefined;
const context = await this.#getTaskRunContext({ run, machine: machine ?? undefined });
return {
id: run.id,
friendlyId: run.friendlyId,
status: run.status,
statusReason: run.statusReason ?? undefined,
createdAt: run.createdAt,
startedAt: run.startedAt,
executedAt: run.executedAt,
updatedAt: run.updatedAt,
delayUntil: run.delayUntil,
expiredAt: run.expiredAt,
completedAt: run.completedAt,
logsDeletedAt: run.logsDeletedAt,
ttl: run.ttl,
taskIdentifier: run.taskIdentifier,
version: run.lockedToVersion?.version,
sdkVersion: run.lockedToVersion?.sdkVersion,
runtime: run.lockedToVersion?.runtime,
runtimeVersion: run.lockedToVersion?.runtimeVersion,
isTest: run.isTest,
replayedFromTaskRunFriendlyId: run.replayedFromTaskRunFriendlyId,
environmentId: run.runtimeEnvironment.id,
idempotencyKey: run.idempotencyKey,
idempotencyKeyExpiresAt: run.idempotencyKeyExpiresAt,
schedule: await this.resolveSchedule(run.scheduleId ?? undefined),
queue: {
name: run.queue,
isCustomQueue: !run.queue.startsWith("task/"),
concurrencyKey: run.concurrencyKey,
},
tags: run.runTags,
baseCostInCents: run.baseCostInCents,
costInCents: run.costInCents,
totalCostInCents: run.costInCents + run.baseCostInCents,
usageDurationMs: run.usageDurationMs,
isFinished,
isRunning: RUNNING_STATUSES.includes(run.status),
isError: isFailedRunStatus(run.status),
payload,
payloadType: run.payloadType,
output,
outputType: run?.outputType ?? "application/json",
error,
relationships: {
root: run.rootTaskRun
? {
...run.rootTaskRun,
isParent: run.parentTaskRun?.friendlyId === run.rootTaskRun.friendlyId,
}
: undefined,
parent: run.parentTaskRun ?? undefined,
},
context: JSON.stringify(context, null, 2),
metadata,
maxDurationInSeconds: getMaxDuration(run.maxDurationInSeconds),
batch: run.batch ? { friendlyId: run.batch.friendlyId } : undefined,
engine: run.engine,
workerQueue: run.workerQueue,
spanId: run.spanId,
isCached: !!span.originalRun,
machinePreset: machine?.name,
};
}
async resolveSchedule(scheduleId?: string) {
if (!scheduleId) {
return;
}
const schedule = await this._replica.taskSchedule.findFirst({
where: {
id: scheduleId,
},
select: {
friendlyId: true,
generatorExpression: true,
timezone: true,
generatorDescription: true,
},
});
if (!schedule) {
return;
}
return {
friendlyId: schedule.friendlyId,
generatorExpression: schedule.generatorExpression,
description: schedule.generatorDescription,
timezone: schedule.timezone,
};
}
async findRun({ span, spanId }: { span: GetSpanResult; spanId: string }) {
const run = await this._replica.taskRun.findFirst({
select: {
id: true,
@@ -132,11 +279,7 @@ export class SpanPresenter extends BasePresenter {
isTest: true,
maxDurationInSeconds: true,
taskEventStore: true,
tags: {
select: {
name: true,
},
},
runTags: true,
machinePreset: true,
lockedToVersion: {
select: {
@@ -219,6 +362,18 @@ export class SpanPresenter extends BasePresenter {
},
},
replayedFromTaskRunFriendlyId: true,
attempts: {
take: 1,
orderBy: {
createdAt: "desc",
},
select: {
number: true,
status: true,
createdAt: true,
friendlyId: true,
},
},
},
where: span.originalRun
? {
@@ -229,180 +384,7 @@ export class SpanPresenter extends BasePresenter {
},
});
if (!run) {
return;
}
const isFinished = isFinalRunStatus(run.status);
const output = !isFinished
? undefined
: run.outputType === "application/store"
? `/resources/packets/${run.runtimeEnvironment.id}/${run.output}`
: typeof run.output !== "undefined" && run.output !== null
? await prettyPrintPacket(run.output, run.outputType ?? undefined)
: undefined;
const payload =
run.payloadType === "application/store"
? `/resources/packets/${run.runtimeEnvironment.id}/${run.payload}`
: typeof run.payload !== "undefined" && run.payload !== null
? await prettyPrintPacket(run.payload, run.payloadType ?? undefined)
: undefined;
let error: TaskRunError | undefined = undefined;
if (run?.error) {
const result = TaskRunError.safeParse(run.error);
if (result.success) {
error = result.data;
} else {
error = {
type: "CUSTOM_ERROR",
raw: JSON.stringify(run.error),
};
}
}
const metadata = run.metadata
? await prettyPrintPacket(run.metadata, run.metadataType, {
filteredKeys: ["$$streams", "$$streamsVersion", "$$streamsBaseUrl"],
})
: undefined;
const machine = run.machinePreset ? machinePresetFromRun(run) : undefined;
const context = {
task: {
id: run.taskIdentifier,
filePath: run.lockedBy?.filePath,
},
run: {
id: run.friendlyId,
createdAt: run.createdAt,
tags: run.tags.map((tag) => tag.name),
isTest: run.isTest,
idempotencyKey: run.idempotencyKey ?? undefined,
startedAt: run.startedAt ?? run.createdAt,
durationMs: run.usageDurationMs,
costInCents: run.costInCents,
baseCostInCents: run.baseCostInCents,
maxAttempts: run.maxAttempts ?? undefined,
version: run.lockedToVersion?.version,
maxDuration: run.maxDurationInSeconds ?? undefined,
},
queue: {
name: run.queue,
},
environment: {
id: run.runtimeEnvironment.id,
slug: run.runtimeEnvironment.slug,
type: run.runtimeEnvironment.type,
},
organization: {
id: run.project.organization.id,
slug: run.project.organization.slug,
name: run.project.organization.title,
},
project: {
id: run.project.id,
ref: run.project.externalRef,
slug: run.project.slug,
name: run.project.name,
},
machine,
};
return {
id: run.id,
friendlyId: run.friendlyId,
status: run.status,
statusReason: run.statusReason ?? undefined,
createdAt: run.createdAt,
startedAt: run.startedAt,
executedAt: run.executedAt,
updatedAt: run.updatedAt,
delayUntil: run.delayUntil,
expiredAt: run.expiredAt,
completedAt: run.completedAt,
logsDeletedAt: run.logsDeletedAt,
ttl: run.ttl,
taskIdentifier: run.taskIdentifier,
version: run.lockedToVersion?.version,
sdkVersion: run.lockedToVersion?.sdkVersion,
runtime: run.lockedToVersion?.runtime,
runtimeVersion: run.lockedToVersion?.runtimeVersion,
isTest: run.isTest,
replayedFromTaskRunFriendlyId: run.replayedFromTaskRunFriendlyId,
environmentId: run.runtimeEnvironment.id,
idempotencyKey: run.idempotencyKey,
idempotencyKeyExpiresAt: run.idempotencyKeyExpiresAt,
schedule: await this.resolveSchedule(run.scheduleId ?? undefined),
queue: {
name: run.queue,
isCustomQueue: !run.queue.startsWith("task/"),
concurrencyKey: run.concurrencyKey,
},
tags: run.tags.map((tag) => tag.name),
baseCostInCents: run.baseCostInCents,
costInCents: run.costInCents,
totalCostInCents: run.costInCents + run.baseCostInCents,
usageDurationMs: run.usageDurationMs,
isFinished,
isRunning: RUNNING_STATUSES.includes(run.status),
isError: isFailedRunStatus(run.status),
payload,
payloadType: run.payloadType,
output,
outputType: run?.outputType ?? "application/json",
error,
relationships: {
root: run.rootTaskRun
? {
...run.rootTaskRun,
isParent: run.parentTaskRun?.friendlyId === run.rootTaskRun.friendlyId,
}
: undefined,
parent: run.parentTaskRun ?? undefined,
},
context: JSON.stringify(context, null, 2),
metadata,
maxDurationInSeconds: getMaxDuration(run.maxDurationInSeconds),
batch: run.batch ? { friendlyId: run.batch.friendlyId } : undefined,
engine: run.engine,
workerQueue: run.workerQueue,
spanId: run.spanId,
isCached: !!span.originalRun,
machinePreset: machine?.name,
};
}
async resolveSchedule(scheduleId?: string) {
if (!scheduleId) {
return;
}
const schedule = await this._replica.taskSchedule.findFirst({
where: {
id: scheduleId,
},
select: {
friendlyId: true,
generatorExpression: true,
timezone: true,
generatorDescription: true,
},
});
if (!schedule) {
return;
}
return {
friendlyId: schedule.friendlyId,
generatorExpression: schedule.generatorExpression,
description: schedule.generatorDescription,
timezone: schedule.timezone,
};
return run;
}
async #getSpan({
@@ -513,4 +495,83 @@ export class SpanPresenter extends BasePresenter {
return { ...data, entity: null };
}
}
async #getTaskRunContext({ run, machine }: { run: FindRunResult; machine?: MachinePreset }) {
if (run.engine === "V1") {
return this.#getV3TaskRunContext({ run, machine });
} else {
return this.#getV4TaskRunContext({ run });
}
}
async #getV3TaskRunContext({
run,
machine,
}: {
run: FindRunResult;
machine?: MachinePreset;
}): Promise<V3TaskRunContext> {
const attempt = run.attempts[0];
const context = {
attempt: attempt
? {
id: attempt.friendlyId,
number: attempt.number,
status: attempt.status,
startedAt: attempt.createdAt,
}
: {
id: AttemptId.generate().friendlyId,
number: 1,
status: "PENDING" as const,
startedAt: run.updatedAt,
},
task: {
id: run.taskIdentifier,
filePath: run.lockedBy?.filePath ?? "",
},
run: {
id: run.friendlyId,
createdAt: run.createdAt,
tags: run.runTags,
isTest: run.isTest,
idempotencyKey: run.idempotencyKey ?? undefined,
startedAt: run.startedAt ?? run.createdAt,
durationMs: run.usageDurationMs,
costInCents: run.costInCents,
baseCostInCents: run.baseCostInCents,
maxAttempts: run.maxAttempts ?? undefined,
version: run.lockedToVersion?.version,
maxDuration: run.maxDurationInSeconds ?? undefined,
},
queue: {
name: run.queue,
id: run.queue,
},
environment: {
id: run.runtimeEnvironment.id,
slug: run.runtimeEnvironment.slug,
type: run.runtimeEnvironment.type,
},
organization: {
id: run.project.organization.id,
slug: run.project.organization.slug,
name: run.project.organization.title,
},
project: {
id: run.project.id,
ref: run.project.externalRef,
slug: run.project.slug,
name: run.project.name,
},
machine,
} satisfies V3TaskRunContext;
return context;
}
async #getV4TaskRunContext({ run }: { run: FindRunResult }): Promise<TaskRunContext> {
return engine.resolveTaskRunContext(run.id);
}
}
apps/webapp/app/routes/resources.runs.$runParam.ts
+36
View File
@@ -76,6 +76,30 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
lockedBy: {
select: {
filePath: true,
worker: {
select: {
deployment: {
select: {
friendlyId: true,
shortCode: true,
version: true,
runtime: true,
runtimeVersion: true,
git: true,
},
},
},
},
},
},
parentTaskRun: {
select: {
friendlyId: true,
},
},
rootTaskRun: {
select: {
friendlyId: true,
},
},
},
@@ -163,6 +187,8 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
baseCostInCents: run.baseCostInCents,
maxAttempts: run.maxAttempts ?? undefined,
version: run.lockedToVersion?.version,
parentTaskRunId: run.parentTaskRun?.friendlyId ?? undefined,
rootTaskRunId: run.rootTaskRun?.friendlyId ?? undefined,
},
queue: {
name: run.queue,
@@ -184,6 +210,16 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
name: run.project.name,
},
machine: run.machinePreset ? machinePresetFromRun(run) : undefined,
deployment: run.lockedBy?.worker.deployment
? {
id: run.lockedBy.worker.deployment.friendlyId,
shortCode: run.lockedBy.worker.deployment.shortCode,
version: run.lockedBy.worker.deployment.version,
runtime: run.lockedBy.worker.deployment.runtime,
runtimeVersion: run.lockedBy.worker.deployment.runtimeVersion,
git: run.lockedBy.worker.deployment.git,
}
: undefined,
};
return typedjson({
apps/webapp/app/v3/failedTaskRun.server.ts
+2 -1
View File
@@ -4,6 +4,7 @@ import {
TaskRunExecution,
TaskRunExecutionRetry,
TaskRunFailedExecutionResult,
V3TaskRunExecution,
} from "@trigger.dev/core/v3";
import type { Prisma, TaskRun } from "@trigger.dev/database";
import * as semver from "semver";
@@ -129,7 +130,7 @@ export class FailedTaskRunRetryHelper extends BaseService {
async #getRetriableAttemptExecution(
run: TaskRunWithAttempts,
completion: TaskRunFailedExecutionResult
): Promise<TaskRunExecution | undefined> {
): Promise<V3TaskRunExecution | undefined> {
let attempt = run.attempts[0];
// We need to create an attempt if:
apps/webapp/app/v3/marqs/devQueueConsumer.server.ts
+2 -2
View File
@@ -1,6 +1,6 @@
import { Context, ROOT_CONTEXT, Span, SpanKind, context, trace } from "@opentelemetry/api";
import {
TaskRunExecution,
V3TaskRunExecution,
TaskRunExecutionLazyAttemptPayload,
TaskRunExecutionResult,
TaskRunFailedExecutionResult,
@@ -138,7 +138,7 @@ export class DevQueueConsumer {
public async taskAttemptCompleted(
workerId: string,
completion: TaskRunExecutionResult,
execution: TaskRunExecution
execution: V3TaskRunExecution
) {
if (completion.ok) {
this._taskSuccesses++;
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
+9 -9
View File
@@ -11,8 +11,8 @@ import {
import {
AckCallbackResult,
MachinePreset,
ProdTaskRunExecution,
ProdTaskRunExecutionPayload,
V3ProdTaskRunExecution,
V3ProdTaskRunExecutionPayload,
TaskRunError,
TaskRunErrorCodes,
TaskRunExecution,
@@ -1679,7 +1679,7 @@ class SharedQueueTasks {
private async _executionFromAttempt(
attempt: AttemptForExecution,
machinePreset?: MachinePreset
): Promise<ProdTaskRunExecution> {
): Promise<V3ProdTaskRunExecution> {
const { backgroundWorkerTask, taskRun, queue } = attempt;
if (!machinePreset) {
@@ -1693,7 +1693,7 @@ class SharedQueueTasks {
dataType: taskRun.metadataType,
});
const execution: ProdTaskRunExecution = {
const execution: V3ProdTaskRunExecution = {
task: {
id: backgroundWorkerTask.slug,
filePath: backgroundWorkerTask.filePath,
@@ -1784,7 +1784,7 @@ class SharedQueueTasks {
setToExecuting?: boolean;
isRetrying?: boolean;
skipStatusChecks?: boolean;
}): Promise<ProdTaskRunExecutionPayload | undefined> {
}): Promise<V3ProdTaskRunExecutionPayload | undefined> {
const attempt = await prisma.taskRunAttempt.findFirst({
where: {
id,
@@ -1874,7 +1874,7 @@ class SharedQueueTasks {
machinePreset
);
const payload: ProdTaskRunExecutionPayload = {
const payload: V3ProdTaskRunExecutionPayload = {
execution,
traceContext: taskRun.traceContext as Record<string, unknown>,
environment: variables.reduce((acc: Record<string, string>, curr) => {
@@ -1888,7 +1888,7 @@ class SharedQueueTasks {
async getResumePayload(attemptId: string): Promise<
| {
execution: ProdTaskRunExecution;
execution: V3ProdTaskRunExecution;
completion: TaskRunExecutionResult;
}
| undefined
@@ -1927,7 +1927,7 @@ class SharedQueueTasks {
async getResumePayloads(attemptIds: string[]): Promise<
Array<{
execution: ProdTaskRunExecution;
execution: V3ProdTaskRunExecution;
completion: TaskRunExecutionResult;
}>
> {
@@ -1985,7 +1985,7 @@ class SharedQueueTasks {
id: string,
setToExecuting?: boolean,
isRetrying?: boolean
): Promise<ProdTaskRunExecutionPayload | undefined> {
): Promise<V3ProdTaskRunExecutionPayload | undefined> {
const run = await prisma.taskRun.findFirst({
where: {
id,
apps/webapp/app/v3/services/completeAttempt.server.ts
+5 -5
View File
@@ -2,13 +2,13 @@ import { Attributes } from "@opentelemetry/api";
import {
MachinePresetName,
TaskRunContext,
TaskRunError,
TaskRunErrorCodes,
TaskRunExecution,
TaskRunExecutionResult,
TaskRunExecutionRetry,
TaskRunFailedExecutionResult,
TaskRunSuccessfulExecutionResult,
V3TaskRunExecution,
flattenAttributes,
isOOMRunError,
sanitizeError,
@@ -60,7 +60,7 @@ export class CompleteAttemptService extends BaseService {
checkpoint,
}: {
completion: TaskRunExecutionResult;
execution: TaskRunExecution;
execution: V3TaskRunExecution;
env?: AuthenticatedEnvironment;
checkpoint?: CheckpointData;
}): Promise<"COMPLETED" | "RETRIED"> {
@@ -196,7 +196,7 @@ export class CompleteAttemptService extends BaseService {
checkpoint,
}: {
completion: TaskRunFailedExecutionResult;
execution: TaskRunExecution;
execution: V3TaskRunExecution;
taskRunAttempt: NonNullable<FoundAttempt>;
env?: AuthenticatedEnvironment;
checkpoint?: CheckpointData;
@@ -559,7 +559,7 @@ export class CompleteAttemptService extends BaseService {
forceRequeue = false,
oomMachine,
}: {
execution: TaskRunExecution;
execution: V3TaskRunExecution;
executionRetry: TaskRunExecutionRetry;
executionRetryInferred: boolean;
taskRunAttempt: NonNullable<FoundAttempt>;
@@ -648,7 +648,7 @@ export class CompleteAttemptService extends BaseService {
executionRetryInferred,
checkpoint,
}: {
execution: TaskRunExecution;
execution: V3TaskRunExecution;
taskRunAttempt: NonNullable<FoundAttempt>;
executionRetry: TaskRunExecutionRetry;
executionRetryInferred: boolean;
apps/webapp/app/v3/services/createTaskRunAttempt.server.ts
+3 -3
View File
@@ -1,4 +1,4 @@
import { parsePacket, TaskRunExecution } from "@trigger.dev/core/v3";
import { parsePacket, V3TaskRunExecution } from "@trigger.dev/core/v3";
import { TaskRun, TaskRunAttempt } from "@trigger.dev/database";
import { MAX_TASK_RUN_ATTEMPTS } from "~/consts";
import { $transaction, prisma, PrismaClientOrTransaction } from "~/db.server";
@@ -25,7 +25,7 @@ export class CreateTaskRunAttemptService extends BaseService {
setToExecuting?: boolean;
startAtZero?: boolean;
}): Promise<{
execution: TaskRunExecution;
execution: V3TaskRunExecution;
run: TaskRun;
attempt: TaskRunAttempt;
}> {
@@ -189,7 +189,7 @@ export class CreateTaskRunAttemptService extends BaseService {
dataType: taskRun.metadataType,
});
const execution: TaskRunExecution = {
const execution: V3TaskRunExecution = {
task: {
id: lockedBy.slug,
filePath: lockedBy.filePath,
internal-packages/cache/README.md
Vendored
+3
View File
@@ -0,0 +1,3 @@
# Redis
This is a simple package that is used to return a valid Redis client and provides an error callback. It will log and swallow errors if they're not handled.
internal-packages/cache/package.json
Vendored
+18
View File
@@ -0,0 +1,18 @@
{
"name": "@internal/cache",
"private": true,
"version": "0.0.1",
"main": "./src/index.ts",
"types": "./src/index.ts",
"type": "module",
"dependencies": {
"@unkey/cache": "^1.5.0",
"@unkey/error": "^0.2.0",
"@trigger.dev/core": "workspace:*",
"@internal/redis": "workspace:*",
"superjson": "^2.2.1"
},
"scripts": {
"typecheck": "tsc --noEmit"
}
}
internal-packages/cache/src/index.ts
Vendored
+8
View File
@@ -0,0 +1,8 @@
export {
createCache,
DefaultStatefulContext,
Namespace,
type Cache as UnkeyCache,
} from "@unkey/cache";
export { MemoryStore } from "@unkey/cache/stores";
export { RedisCacheStore } from "./stores/redis.js";
internal-packages/cache/src/stores/redis.ts
Vendored
+105
View File
@@ -0,0 +1,105 @@
import { CacheError } from "@unkey/cache";
import type { Entry, Store } from "@unkey/cache/stores";
import { Err, Ok, type Result } from "@unkey/error";
import { createRedisClient, Redis, RedisOptions } from "@internal/redis";
export type RedisCacheStoreConfig = {
connection: RedisOptions;
name?: string;
useModernCacheKeyBuilder?: boolean;
};
export class RedisCacheStore<TNamespace extends string, TValue = any>
implements Store<TNamespace, TValue>
{
public readonly name = "redis";
private readonly redis: Redis;
constructor(private readonly config: RedisCacheStoreConfig) {
this.redis = createRedisClient({
...config.connection,
name: config.name ?? "trigger:cacheStore",
});
}
private buildCacheKey(namespace: TNamespace, key: string): string {
if (this.config.useModernCacheKeyBuilder) {
return [namespace, key].join(":");
}
return [namespace, key].join("::");
}
public async get(
namespace: TNamespace,
key: string
): Promise<Result<Entry<TValue> | undefined, CacheError>> {
let raw: string | null;
try {
raw = await this.redis.get(this.buildCacheKey(namespace, key));
} catch (err) {
return Err(
new CacheError({
tier: this.name,
key,
message: (err as Error).message,
})
);
}
if (!raw) {
return Promise.resolve(Ok(undefined));
}
try {
const superjson = await import("superjson");
const entry = superjson.parse(raw) as Entry<TValue>;
return Ok(entry);
} catch (err) {
return Err(
new CacheError({
tier: this.name,
key,
message: (err as Error).message,
})
);
}
}
public async set(
namespace: TNamespace,
key: string,
entry: Entry<TValue>
): Promise<Result<void, CacheError>> {
const cacheKey = this.buildCacheKey(namespace, key);
try {
const superjson = await import("superjson");
await this.redis.set(cacheKey, superjson.stringify(entry), "PXAT", entry.staleUntil);
return Ok();
} catch (err) {
return Err(
new CacheError({
tier: this.name,
key,
message: (err as Error).message,
})
);
}
}
public async remove(namespace: TNamespace, key: string): Promise<Result<void, CacheError>> {
try {
const cacheKey = this.buildCacheKey(namespace, key);
await this.redis.del(cacheKey);
return Promise.resolve(Ok());
} catch (err) {
return Err(
new CacheError({
tier: this.name,
key,
message: (err as Error).message,
})
);
}
}
}
internal-packages/cache/tsconfig.json
Vendored
+23
View File
@@ -0,0 +1,23 @@
{
"compilerOptions": {
"target": "ES2019",
"lib": ["ES2019", "DOM", "DOM.Iterable", "DOM.AsyncIterable"],
"module": "Node16",
"moduleResolution": "Node16",
"moduleDetection": "force",
"verbatimModuleSyntax": false,
"types": ["vitest/globals"],
"esModuleInterop": true,
"forceConsistentCasingInFileNames": true,
"isolatedModules": true,
"preserveWatchOutput": true,
"skipLibCheck": true,
"noEmit": true,
"strict": true,
"paths": {
"@trigger.dev/core": ["../../packages/core/src/index"],
"@trigger.dev/core/*": ["../../packages/core/src/*"]
}
},
"exclude": ["node_modules"]
}
internal-packages/run-engine/package.json
+1 -1
View File
@@ -25,7 +25,7 @@
"@internal/tracing": "workspace:*",
"@trigger.dev/core": "workspace:*",
"@trigger.dev/database": "workspace:*",
"@unkey/cache": "^1.5.0",
"@internal/cache": "workspace:*",
"assert-never": "^1.2.1",
"nanoid": "3.3.8",
"redlock": "5.0.0-beta.2",
internal-packages/run-engine/src/engine/index.ts
+6
View File
@@ -9,6 +9,7 @@ import {
ExecutionResult,
RunExecutionData,
StartRunAttemptResult,
TaskRunContext,
TaskRunExecutionResult,
} from "@trigger.dev/core/v3";
import { RunId, WaitpointId } from "@trigger.dev/core/v3/isomorphic";
@@ -288,6 +289,7 @@ export class RunEngine {
delayedRunSystem: this.delayedRunSystem,
machines: this.options.machines,
retryWarmStartThresholdMs: this.options.retryWarmStartThresholdMs,
redisOptions: this.options.cache?.redis ?? this.options.runLock.redis,
});
this.dequeueSystem = new DequeueSystem({
@@ -1054,6 +1056,10 @@ export class RunEngine {
}
}
async resolveTaskRunContext(runId: string): Promise<TaskRunContext> {
return this.runAttemptSystem.resolveTaskRunContext(runId);
}
async getSnapshotsSince({
runId,
snapshotId,
internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts
+598 -136
View File
@@ -1,3 +1,12 @@
import {
createCache,
DefaultStatefulContext,
MemoryStore,
Namespace,
RedisCacheStore,
UnkeyCache,
} from "@internal/cache";
import { RedisOptions } from "@internal/redis";
import { startSpan } from "@internal/tracing";
import { tryCatch } from "@trigger.dev/core/utils";
import {
@@ -5,9 +14,16 @@ import {
ExecutionResult,
FlushedRunMetadata,
GitMeta,
MachinePreset,
MachinePresetName,
StartRunAttemptResult,
TaskRunContext,
TaskRunError,
TaskRunExecution,
TaskRunExecutionDeployment,
TaskRunExecutionOrganization,
TaskRunExecutionProject,
TaskRunExecutionQueue,
TaskRunExecutionResult,
TaskRunFailedExecutionResult,
TaskRunInternalError,
@@ -23,7 +39,7 @@ import {
import { MAX_TASK_RUN_ATTEMPTS } from "../consts.js";
import { runStatusFromError, ServiceValidationError } from "../errors.js";
import { sendNotificationToWorker } from "../eventBus.js";
import { getMachinePreset } from "../machinePresets.js";
import { getMachinePreset, machinePresetFromName } from "../machinePresets.js";
import { retryOutcomeFromCompletion } from "../retrying.js";
import { isExecuting, isInitialState } from "../statuses.js";
import { RunEngineOptions } from "../types.js";
@@ -36,6 +52,7 @@ import {
} from "./executionSnapshotSystem.js";
import { SystemResources } from "./systems.js";
import { WaitpointSystem } from "./waitpointSystem.js";
import { BatchId, RunId } from "@trigger.dev/core/v3/isomorphic";
export type RunAttemptSystemOptions = {
resources: SystemResources;
@@ -45,14 +62,54 @@ export type RunAttemptSystemOptions = {
delayedRunSystem: DelayedRunSystem;
retryWarmStartThresholdMs?: number;
machines: RunEngineOptions["machines"];
redisOptions: RedisOptions;
};
type BackwardsCompatibleTaskRunExecution = Omit<TaskRunExecution, "task" | "attempt" | "run"> & {
task: TaskRunExecution["task"] & {
exportName: string | undefined;
};
attempt: TaskRunExecution["attempt"] & {
id: string;
backgroundWorkerId: string;
backgroundWorkerTaskId: string;
status: string;
};
run: TaskRunExecution["run"] & {
context: undefined;
durationMs: number;
costInCents: number;
baseCostInCents: number;
};
};
const ORG_FRESH_TTL = 60000 * 60 * 24; // 1 day
const ORG_STALE_TTL = 60000 * 60 * 24 * 2; // 2 days
const PROJECT_FRESH_TTL = 60000 * 60 * 24; // 1 day
const PROJECT_STALE_TTL = 60000 * 60 * 24 * 2; // 2 days
const TASK_FRESH_TTL = 60000 * 60 * 24; // 1 day
const TASK_STALE_TTL = 60000 * 60 * 24 * 2; // 2 days
const MACHINE_PRESET_FRESH_TTL = 60000 * 60 * 24; // 1 day
const MACHINE_PRESET_STALE_TTL = 60000 * 60 * 24 * 2; // 2 days
const DEPLOYMENT_FRESH_TTL = 60000 * 60 * 24; // 1 day
const DEPLOYMENT_STALE_TTL = 60000 * 60 * 24 * 2; // 2 days
const QUEUE_FRESH_TTL = 60000 * 60; // 1 hour
const QUEUE_STALE_TTL = 60000 * 60 * 2; // 2 hours
export class RunAttemptSystem {
private readonly $: SystemResources;
private readonly executionSnapshotSystem: ExecutionSnapshotSystem;
private readonly batchSystem: BatchSystem;
private readonly waitpointSystem: WaitpointSystem;
private readonly delayedRunSystem: DelayedRunSystem;
private readonly cache: UnkeyCache<{
tasks: BackwardsCompatibleTaskRunExecution["task"];
machinePresets: MachinePreset;
deployments: TaskRunExecutionDeployment;
queues: TaskRunExecutionQueue;
projects: TaskRunExecutionProject;
orgs: TaskRunExecutionOrganization;
}>;
constructor(private readonly options: RunAttemptSystemOptions) {
this.$ = options.resources;
@@ -60,6 +117,170 @@ export class RunAttemptSystem {
this.batchSystem = options.batchSystem;
this.waitpointSystem = options.waitpointSystem;
this.delayedRunSystem = options.delayedRunSystem;
const ctx = new DefaultStatefulContext();
// TODO: use an LRU cache for memory store
const memory = new MemoryStore({ persistentMap: new Map() });
const redisCacheStore = new RedisCacheStore({
name: "run-attempt-system",
connection: {
...options.redisOptions,
keyPrefix: "engine:run-attempt-system:cache:",
},
useModernCacheKeyBuilder: true,
});
this.cache = createCache({
orgs: new Namespace<TaskRunExecutionOrganization>(ctx, {
stores: [memory, redisCacheStore],
fresh: ORG_FRESH_TTL,
stale: ORG_STALE_TTL,
}),
projects: new Namespace<TaskRunExecutionProject>(ctx, {
stores: [memory, redisCacheStore],
fresh: PROJECT_FRESH_TTL,
stale: PROJECT_STALE_TTL,
}),
tasks: new Namespace<BackwardsCompatibleTaskRunExecution["task"]>(ctx, {
stores: [memory, redisCacheStore],
fresh: TASK_FRESH_TTL,
stale: TASK_STALE_TTL,
}),
machinePresets: new Namespace<MachinePreset>(ctx, {
stores: [memory, redisCacheStore],
fresh: MACHINE_PRESET_FRESH_TTL,
stale: MACHINE_PRESET_STALE_TTL,
}),
deployments: new Namespace<TaskRunExecutionDeployment>(ctx, {
stores: [memory, redisCacheStore],
fresh: DEPLOYMENT_FRESH_TTL,
stale: DEPLOYMENT_STALE_TTL,
}),
queues: new Namespace<TaskRunExecutionQueue>(ctx, {
stores: [memory, redisCacheStore],
fresh: QUEUE_FRESH_TTL,
stale: QUEUE_STALE_TTL,
}),
});
}
public async resolveTaskRunContext(runId: string): Promise<TaskRunContext> {
const run = await this.$.prisma.taskRun.findFirst({
where: {
id: runId,
},
select: {
id: true,
createdAt: true,
updatedAt: true,
executedAt: true,
baseCostInCents: true,
projectId: true,
organizationId: true,
friendlyId: true,
lockedById: true,
lockedQueueId: true,
queue: true,
attemptNumber: true,
status: true,
ttl: true,
machinePreset: true,
runTags: true,
isTest: true,
idempotencyKey: true,
startedAt: true,
maxAttempts: true,
taskVersion: true,
maxDurationInSeconds: true,
usageDurationMs: true,
costInCents: true,
traceContext: true,
priorityMs: true,
taskIdentifier: true,
runtimeEnvironment: {
select: {
id: true,
slug: true,
type: true,
branchName: true,
git: true,
organizationId: true,
},
},
parentTaskRunId: true,
rootTaskRunId: true,
batchId: true,
},
});
if (!run) {
throw new ServiceValidationError("Task run not found", 404);
}
const [task, queue, organization, project, machinePreset, deployment] = await Promise.all([
run.lockedById
? this.#resolveTaskRunExecutionTask(run.lockedById)
: Promise.resolve({
id: run.taskIdentifier,
filePath: "unknown",
}),
this.#resolveTaskRunExecutionQueue({
runId,
lockedQueueId: run.lockedQueueId ?? undefined,
queueName: run.queue,
runtimeEnvironmentId: run.runtimeEnvironment.id,
}),
this.#resolveTaskRunExecutionOrganization(run.runtimeEnvironment.organizationId),
this.#resolveTaskRunExecutionProjectByRuntimeEnvironmentId(run.runtimeEnvironment.id),
run.lockedById
? this.#resolveTaskRunExecutionMachinePreset(run.lockedById, run.machinePreset)
: Promise.resolve(
getMachinePreset({
defaultMachine: this.options.machines.defaultMachine,
machines: this.options.machines.machines,
config: undefined,
run,
})
),
run.lockedById
? this.#resolveTaskRunExecutionDeployment(run.lockedById)
: Promise.resolve(undefined),
]);
return {
run: {
id: run.friendlyId,
tags: run.runTags,
isTest: run.isTest,
createdAt: run.createdAt,
startedAt: run.startedAt ?? run.createdAt,
idempotencyKey: run.idempotencyKey ?? undefined,
maxAttempts: run.maxAttempts ?? undefined,
version: run.taskVersion ?? "unknown",
maxDuration: run.maxDurationInSeconds ?? undefined,
priority: run.priorityMs === 0 ? undefined : run.priorityMs / 1_000,
parentTaskRunId: run.parentTaskRunId ? RunId.toFriendlyId(run.parentTaskRunId) : undefined,
rootTaskRunId: run.rootTaskRunId ? RunId.toFriendlyId(run.rootTaskRunId) : undefined,
},
attempt: {
number: run.attemptNumber ?? 1,
startedAt: run.startedAt ?? new Date(),
},
task,
queue,
organization,
project,
machine: machinePreset,
deployment,
environment: {
id: run.runtimeEnvironment.id,
slug: run.runtimeEnvironment.slug,
type: run.runtimeEnvironment.type,
branchName: run.runtimeEnvironment.branchName ?? undefined,
git: safeParseGitMeta(run.runtimeEnvironment.git),
},
batch: run.batchId ? { id: BatchId.toFriendlyId(run.batchId) } : undefined,
};
}
public async startRunAttempt({
@@ -95,35 +316,19 @@ export class RunAttemptSystem {
throw new ServiceValidationError("Snapshot changed", 409);
}
const environment = await this.#getAuthenticatedEnvironmentFromRun(runId, prisma);
if (!environment) {
throw new ServiceValidationError("Environment not found", 404);
}
const taskRun = await prisma.taskRun.findFirst({
where: {
id: runId,
},
include: {
tags: true,
lockedBy: {
include: {
worker: {
select: {
id: true,
version: true,
sdkVersion: true,
cliVersion: true,
supportsLazyAttempts: true,
},
},
},
},
batchItems: {
include: {
batchTaskRun: true,
},
},
select: {
id: true,
friendlyId: true,
attemptNumber: true,
projectId: true,
runtimeEnvironmentId: true,
status: true,
lockedById: true,
ttl: true,
},
});
@@ -142,21 +347,10 @@ export class RunAttemptSystem {
throw new ServiceValidationError("Task run is cancelled", 400);
}
if (!taskRun.lockedBy) {
if (!taskRun.lockedById) {
throw new ServiceValidationError("Task run is not locked", 400);
}
const queue = await prisma.taskQueue.findFirst({
where: {
runtimeEnvironmentId: environment.id,
name: taskRun.queue,
},
});
if (!queue) {
throw new ServiceValidationError("Queue not found", 404);
}
//increment the attempt number (start at 1)
const nextAttemptNumber = (taskRun.attemptNumber ?? 0) + 1;
@@ -190,11 +384,50 @@ export class RunAttemptSystem {
attemptNumber: nextAttemptNumber,
executedAt: taskRun.attemptNumber === null ? new Date() : undefined,
},
include: {
tags: true,
lockedBy: {
include: { worker: true },
select: {
id: true,
createdAt: true,
updatedAt: true,
executedAt: true,
baseCostInCents: true,
projectId: true,
organizationId: true,
friendlyId: true,
lockedById: true,
lockedQueueId: true,
queue: true,
attemptNumber: true,
status: true,
ttl: true,
metadata: true,
metadataType: true,
machinePreset: true,
payload: true,
payloadType: true,
runTags: true,
isTest: true,
idempotencyKey: true,
startedAt: true,
maxAttempts: true,
taskVersion: true,
maxDurationInSeconds: true,
usageDurationMs: true,
costInCents: true,
traceContext: true,
priorityMs: true,
batchId: true,
runtimeEnvironment: {
select: {
id: true,
slug: true,
type: true,
branchName: true,
git: true,
organizationId: true,
},
},
parentTaskRunId: true,
rootTaskRunId: true,
},
});
@@ -222,7 +455,7 @@ export class RunAttemptSystem {
await this.$.worker.ack(`expireRun:${taskRun.id}`);
}
return { run, snapshot: newSnapshot };
return { updatedRun: run, snapshot: newSnapshot };
},
(error) => {
this.$.logger.error("RunEngine.createRunAttempt(): prisma.$transaction error", {
@@ -247,56 +480,59 @@ export class RunAttemptSystem {
throw new ServiceValidationError("Failed to create task run attempt", 500);
}
const { run, snapshot } = result;
const { updatedRun, snapshot } = result;
this.$.eventBus.emit("runAttemptStarted", {
time: new Date(),
run: {
id: run.id,
status: run.status,
createdAt: run.createdAt,
updatedAt: run.updatedAt,
id: updatedRun.id,
status: updatedRun.status,
createdAt: updatedRun.createdAt,
updatedAt: updatedRun.updatedAt,
attemptNumber: nextAttemptNumber,
baseCostInCents: run.baseCostInCents,
executedAt: run.executedAt ?? undefined,
baseCostInCents: updatedRun.baseCostInCents,
executedAt: updatedRun.executedAt ?? undefined,
},
organization: {
id: environment.organization.id,
id: updatedRun.runtimeEnvironment.organizationId,
},
project: {
id: environment.project.id,
id: updatedRun.projectId,
},
environment: {
id: environment.id,
id: updatedRun.runtimeEnvironment.id,
},
});
const machinePreset = getMachinePreset({
machines: this.options.machines.machines,
defaultMachine: this.options.machines.defaultMachine,
config: taskRun.lockedBy.machineConfig ?? {},
run: taskRun,
});
const environmentGit = safeParseGitMeta(updatedRun.runtimeEnvironment.git);
const metadata = await parsePacket({
data: taskRun.metadata ?? undefined,
dataType: taskRun.metadataType,
});
const [metadata, task, queue, organization, project, machinePreset, deployment] =
await Promise.all([
parsePacket({
data: updatedRun.metadata ?? undefined,
dataType: updatedRun.metadataType,
}),
this.#resolveTaskRunExecutionTask(taskRun.lockedById),
this.#resolveTaskRunExecutionQueue({
runId,
lockedQueueId: updatedRun.lockedQueueId ?? undefined,
queueName: updatedRun.queue,
runtimeEnvironmentId: updatedRun.runtimeEnvironment.id,
}),
this.#resolveTaskRunExecutionOrganization(
updatedRun.runtimeEnvironment.organizationId
),
this.#resolveTaskRunExecutionProjectByRuntimeEnvironmentId(
updatedRun.runtimeEnvironment.id
),
this.#resolveTaskRunExecutionMachinePreset(
taskRun.lockedById,
updatedRun.machinePreset
),
this.#resolveTaskRunExecutionDeployment(taskRun.lockedById),
]);
let git: GitMeta | undefined = undefined;
if (environment.git) {
const parsed = GitMeta.safeParse(environment.git);
if (parsed.success) {
git = parsed.data;
}
}
const execution: TaskRunExecution = {
task: {
id: run.lockedBy!.slug,
filePath: run.lockedBy!.filePath,
exportName: run.lockedBy!.exportName ?? undefined,
},
const execution: BackwardsCompatibleTaskRunExecution = {
attempt: {
number: nextAttemptNumber,
startedAt: latestSnapshot.updatedAt,
@@ -310,59 +546,56 @@ export class RunAttemptSystem {
status: "deprecated",
},
run: {
id: run.friendlyId,
payload: run.payload,
payloadType: run.payloadType,
createdAt: run.createdAt,
tags: run.tags.map((tag) => tag.name),
isTest: run.isTest,
idempotencyKey: run.idempotencyKey ?? undefined,
startedAt: run.startedAt ?? run.createdAt,
maxAttempts: run.maxAttempts ?? undefined,
version: run.lockedBy!.worker.version,
id: updatedRun.friendlyId,
payload: updatedRun.payload,
payloadType: updatedRun.payloadType,
createdAt: updatedRun.createdAt,
tags: updatedRun.runTags,
isTest: updatedRun.isTest,
idempotencyKey: updatedRun.idempotencyKey ?? undefined,
startedAt: updatedRun.startedAt ?? updatedRun.createdAt,
maxAttempts: updatedRun.maxAttempts ?? undefined,
version: updatedRun.taskVersion ?? "unknown",
metadata,
maxDuration: run.maxDurationInSeconds ?? undefined,
maxDuration: updatedRun.maxDurationInSeconds ?? undefined,
/** @deprecated */
context: undefined,
/** @deprecated */
durationMs: run.usageDurationMs,
durationMs: updatedRun.usageDurationMs,
/** @deprecated */
costInCents: run.costInCents,
costInCents: updatedRun.costInCents,
/** @deprecated */
baseCostInCents: run.baseCostInCents,
traceContext: run.traceContext as Record<string, string | undefined>,
priority: run.priorityMs === 0 ? undefined : run.priorityMs / 1_000,
},
queue: {
id: queue.friendlyId,
name: queue.name,
},
environment: {
id: environment.id,
slug: environment.slug,
type: environment.type,
branchName: environment.branchName ?? undefined,
git,
},
organization: {
id: environment.organization.id,
slug: environment.organization.slug,
name: environment.organization.title,
},
project: {
id: environment.project.id,
ref: environment.project.externalRef,
slug: environment.project.slug,
name: environment.project.name,
},
batch:
taskRun.batchItems[0] && taskRun.batchItems[0].batchTaskRun
? { id: taskRun.batchItems[0].batchTaskRun.friendlyId }
baseCostInCents: updatedRun.baseCostInCents,
traceContext: updatedRun.traceContext as Record<string, string | undefined>,
priority: updatedRun.priorityMs === 0 ? undefined : updatedRun.priorityMs / 1_000,
parentTaskRunId: updatedRun.parentTaskRunId
? RunId.toFriendlyId(updatedRun.parentTaskRunId)
: undefined,
rootTaskRunId: updatedRun.rootTaskRunId
? RunId.toFriendlyId(updatedRun.rootTaskRunId)
: undefined,
},
task,
queue,
environment: {
id: updatedRun.runtimeEnvironment.id,
slug: updatedRun.runtimeEnvironment.slug,
type: updatedRun.runtimeEnvironment.type,
branchName: updatedRun.runtimeEnvironment.branchName ?? undefined,
git: environmentGit,
},
organization,
project,
machine: machinePreset,
deployment,
batch: updatedRun.batchId
? {
id: BatchId.toFriendlyId(updatedRun.batchId),
}
: undefined,
};
return { run, snapshot, execution };
return { run: updatedRun, snapshot, execution };
});
},
{
@@ -1311,27 +1544,248 @@ export class RunAttemptSystem {
await this.$.worker.ack(`heartbeatSnapshot.${id}`);
}
async #getAuthenticatedEnvironmentFromRun(runId: string, tx?: PrismaClientOrTransaction) {
const prisma = tx ?? this.$.prisma;
const taskRun = await prisma.taskRun.findFirst({
where: {
id: runId,
},
include: {
runtimeEnvironment: {
include: {
organization: true,
project: true,
},
async #resolveTaskRunExecutionTask(
backgroundWorkerTaskId: string
): Promise<BackwardsCompatibleTaskRunExecution["task"]> {
const result = await this.cache.tasks.swr(backgroundWorkerTaskId, async () => {
const task = await this.$.prisma.backgroundWorkerTask.findFirstOrThrow({
where: {
id: backgroundWorkerTaskId,
},
},
select: {
id: true,
slug: true,
filePath: true,
exportName: true,
},
});
return {
id: task.slug,
filePath: task.filePath,
exportName: task.exportName ?? undefined,
};
});
if (!taskRun) {
return;
if (result.err) {
throw result.err;
}
return taskRun?.runtimeEnvironment;
if (!result.val) {
throw new ServiceValidationError(
`Could not resolve task execution data for task ${backgroundWorkerTaskId}`
);
}
return result.val;
}
async #resolveTaskRunExecutionOrganization(
organizationId: string
): Promise<TaskRunExecutionOrganization> {
const result = await this.cache.orgs.swr(organizationId, async () => {
const organization = await this.$.prisma.organization.findFirstOrThrow({
where: { id: organizationId },
select: {
id: true,
title: true,
slug: true,
},
});
return {
id: organization.id,
name: organization.title,
slug: organization.slug,
};
});
if (result.err) {
throw result.err;
}
if (!result.val) {
throw new ServiceValidationError(
`Could not resolve organization data for organization ${organizationId}`
);
}
return result.val;
}
async #resolveTaskRunExecutionProjectByRuntimeEnvironmentId(
runtimeEnvironmentId: string
): Promise<TaskRunExecutionProject> {
const result = await this.cache.projects.swr(runtimeEnvironmentId, async () => {
const { project } = await this.$.prisma.runtimeEnvironment.findFirstOrThrow({
where: { id: runtimeEnvironmentId },
select: {
id: true,
project: {
select: {
id: true,
name: true,
slug: true,
externalRef: true,
},
},
},
});
return {
id: project.id,
name: project.name,
slug: project.slug,
ref: project.externalRef,
};
});
if (result.err) {
throw result.err;
}
if (!result.val) {
throw new ServiceValidationError(
`Could not resolve project data for project ${runtimeEnvironmentId}`
);
}
return result.val;
}
async #resolveTaskRunExecutionMachinePreset(
backgroundWorkerTaskId: string,
runMachinePreset: string | null
): Promise<MachinePreset> {
if (runMachinePreset) {
return machinePresetFromName(
this.options.machines.machines,
runMachinePreset as MachinePresetName
);
}
const result = await this.cache.machinePresets.swr(backgroundWorkerTaskId, async () => {
const { machineConfig } = await this.$.prisma.backgroundWorkerTask.findFirstOrThrow({
where: {
id: backgroundWorkerTaskId,
},
select: {
machineConfig: true,
},
});
return getMachinePreset({
machines: this.options.machines.machines,
defaultMachine: this.options.machines.defaultMachine,
config: machineConfig,
run: { machinePreset: null },
});
});
if (result.err) {
throw result.err;
}
if (!result.val) {
throw new ServiceValidationError(
`Could not resolve machine preset for task ${backgroundWorkerTaskId}`
);
}
return result.val;
}
async #resolveTaskRunExecutionQueue(params: {
runId: string;
lockedQueueId?: string;
queueName: string;
runtimeEnvironmentId: string;
}): Promise<TaskRunExecutionQueue> {
const result = await this.cache.queues.swr(params.runId, async () => {
const queue = params.lockedQueueId
? await this.$.prisma.taskQueue.findFirst({
where: {
id: params.lockedQueueId,
},
select: {
id: true,
friendlyId: true,
name: true,
},
})
: await this.$.prisma.taskQueue.findFirst({
where: {
runtimeEnvironmentId: params.runtimeEnvironmentId,
name: params.queueName,
},
select: {
id: true,
friendlyId: true,
name: true,
},
});
if (!queue) {
throw new ServiceValidationError(
`Could not resolve queue data for queue ${params.queueName}`,
404
);
}
return {
id: queue.friendlyId,
name: queue.name,
};
});
if (result.err) {
throw result.err;
}
if (!result.val) {
throw new ServiceValidationError(
`Could not resolve queue data for queue ${params.queueName}`,
404
);
}
return result.val;
}
async #resolveTaskRunExecutionDeployment(
backgroundWorkerTaskId: string
): Promise<TaskRunExecutionDeployment | undefined> {
const result = await this.cache.deployments.swr(backgroundWorkerTaskId, async () => {
const { worker } = await this.$.prisma.backgroundWorkerTask.findFirstOrThrow({
where: { id: backgroundWorkerTaskId },
select: {
worker: {
select: {
deployment: true,
},
},
},
});
if (!worker.deployment) {
return undefined;
}
return {
id: worker.deployment.friendlyId,
shortCode: worker.deployment.shortCode,
version: worker.deployment.version,
runtime: worker.deployment.runtime ?? "unknown",
runtimeVersion: worker.deployment.runtimeVersion ?? "unknown",
git: safeParseGitMeta(worker.deployment.git),
};
});
if (result.err) {
throw result.err;
}
return result.val;
}
async #notifyMetadataUpdated(runId: string, completion: TaskRunExecutionResult) {
@@ -1386,3 +1840,11 @@ export class RunAttemptSystem {
}
}
}
export function safeParseGitMeta(git: unknown): GitMeta | undefined {
const parsed = GitMeta.safeParse(git);
if (parsed.success) {
return parsed.data;
}
return undefined;
}
internal-packages/run-engine/src/engine/tests/heartbeats.test.ts
+6 -7
View File
@@ -639,7 +639,7 @@ describe("RunEngine heartbeats", () => {
containerTest("Heartbeat keeps run alive", async ({ prisma, redisOptions }) => {
const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
const executingTimeout = 100;
const executingTimeout = 500;
const engine = new RunEngine({
prisma,
@@ -726,24 +726,23 @@ describe("RunEngine heartbeats", () => {
expect(executionData.snapshot.executionStatus).toBe("EXECUTING");
expect(executionData.run.status).toBe("EXECUTING");
// Send heartbeats every 50ms (half the timeout)
for (let i = 0; i < 6; i++) {
await setTimeout(50);
// Send heartbeats every 100ms (to make sure we're not timing out)
for (let i = 0; i < 5; i++) {
await setTimeout(100);
await engine.heartbeatRun({
runId: run.id,
snapshotId: attempt.snapshot.id,
});
}
// After 300ms (3x the timeout) the run should still be executing
// because we've been sending heartbeats
// Should still be executing because we're sending heartbeats
const executionData2 = await engine.getRunExecutionData({ runId: run.id });
assertNonNullable(executionData2);
expect(executionData2.snapshot.executionStatus).toBe("EXECUTING");
expect(executionData2.run.status).toBe("EXECUTING");
// Stop sending heartbeats and wait for timeout
await setTimeout(executingTimeout * 3);
await setTimeout(executingTimeout * 2);
// Now it should have timed out and be queued
const executionData3 = await engine.getRunExecutionData({ runId: run.id });
internal-packages/run-engine/src/engine/types.ts
+3
View File
@@ -58,6 +58,9 @@ export type RunEngineOptions = {
automaticExtensionThreshold?: number;
retryConfig?: LockRetryConfig;
};
cache?: {
redis: RedisOptions;
};
/** If not set then checkpoints won't ever be used */
retryWarmStartThresholdMs?: number;
heartbeatTimeoutsMs?: Partial<HeartbeatTimeouts>;
internal-packages/run-engine/src/run-queue/fairQueueSelectionStrategy.ts
+7 -2
View File
@@ -1,7 +1,12 @@
import { createRedisClient, Redis, type RedisOptions } from "@internal/redis";
import { startSpan, type Tracer } from "@internal/tracing";
import { createCache, DefaultStatefulContext, Namespace, Cache as UnkeyCache } from "@unkey/cache";
import { MemoryStore } from "@unkey/cache/stores";
import {
createCache,
DefaultStatefulContext,
Namespace,
type UnkeyCache,
MemoryStore,
} from "@internal/cache";
import { randomUUID } from "crypto";
import seedrandom from "seedrandom";
import {
packages/cli-v3/src/entryPoints/dev-run-worker.ts
+7
View File
@@ -484,6 +484,8 @@ const zodIpc = new ZodIpcConnection({
}
runMetadataManager.runId = execution.run.id;
runMetadataManager.runIdIsRoot = typeof execution.run.rootTaskRunId === "undefined";
_executionCount++;
const executor = new TaskExecutor(task, {
@@ -503,6 +505,11 @@ const zodIpc = new ZodIpcConnection({
getNumberEnvVar("TRIGGER_RUN_METADATA_FLUSH_INTERVAL", 1000)
);
devUsageManager.setInitialState({
cpuTime: execution.run.durationMs ?? 0,
costInCents: execution.run.costInCents ?? 0,
});
_executionMeasurement = usage.start();
const timeoutController = timeout.abortAfterTimeout(execution.run.maxDuration);
packages/cli-v3/src/entryPoints/managed-run-worker.ts
+9 -1
View File
@@ -328,12 +328,17 @@ const zodIpc = new ZodIpcConnection({
resetExecutionEnvironment();
initializeUsageManager({
const prodManager = initializeUsageManager({
usageIntervalMs: getEnvVar("USAGE_HEARTBEAT_INTERVAL_MS"),
usageEventUrl: getEnvVar("USAGE_EVENT_URL"),
triggerJWT: getEnvVar("TRIGGER_JWT"),
});
prodManager.setInitialState({
cpuTime: execution.run.durationMs ?? 0,
costInCents: execution.run.costInCents ?? 0,
});
standardRunTimelineMetricsManager.registerMetricsFromExecution(metrics, isWarmStart);
console.log(`[${new Date().toISOString()}] Received EXECUTE_TASK_RUN`, execution);
@@ -483,6 +488,7 @@ const zodIpc = new ZodIpcConnection({
}
runMetadataManager.runId = execution.run.id;
runMetadataManager.runIdIsRoot = typeof execution.run.rootTaskRunId === "undefined";
_executionCount++;
const executor = new TaskExecutor(task, {
@@ -689,6 +695,8 @@ function initializeUsageManager({
usage.setGlobalUsageManager(prodUsageManager);
timeout.setGlobalManager(new UsageTimeoutManager(devUsageManager));
return prodUsageManager;
}
_sharedWorkerRuntime = new SharedRuntimeManager(zodIpc, true);
packages/core/src/v3/isomorphic/friendlyId.ts
+1
View File
@@ -95,6 +95,7 @@ export const SnapshotId = new IdUtil("snapshot");
export const WaitpointId = new IdUtil("waitpoint");
export const BatchId = new IdUtil("batch");
export const BulkActionId = new IdUtil("bulk");
export const AttemptId = new IdUtil("attempt");
export class IdGenerator {
private alphabet: string;
packages/core/src/v3/runMetadata/manager.ts
+88 -2
View File
@@ -24,6 +24,7 @@ export class StandardMetadataManager implements RunMetadataManager {
private queuedRootOperations: Set<RunMetadataChangeOperation> = new Set();
public runId: string | undefined;
public runIdIsRoot: boolean = false;
constructor(
private apiClient: ApiClient,
@@ -38,6 +39,7 @@ export class StandardMetadataManager implements RunMetadataManager {
this.activeStreams.clear();
this.store = undefined;
this.runId = undefined;
this.runIdIsRoot = false;
if (this.flushTimeoutId) {
clearTimeout(this.flushTimeoutId);
@@ -54,34 +56,76 @@ export class StandardMetadataManager implements RunMetadataManager {
// Create the updater object and store it in a local variable
const parentUpdater: RunMetadataUpdater = {
set: (key, value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.set(key, value);
}
self.queuedParentOperations.add({ type: "set", key, value });
return parentUpdater;
},
del: (key) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.del(key);
}
self.queuedParentOperations.add({ type: "delete", key });
return parentUpdater;
},
append: (key, value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.append(key, value);
}
self.queuedParentOperations.add({ type: "append", key, value });
return parentUpdater;
},
remove: (key, value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.remove(key, value);
}
self.queuedParentOperations.add({ type: "remove", key, value });
return parentUpdater;
},
increment: (key, value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.increment(key, value);
}
self.queuedParentOperations.add({ type: "increment", key, value });
return parentUpdater;
},
decrement: (key, value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.decrement(key, value);
}
self.queuedParentOperations.add({ type: "increment", key, value: -Math.abs(value) });
return parentUpdater;
},
update: (value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.update(value);
}
self.queuedParentOperations.add({ type: "update", value });
return parentUpdater;
},
stream: (key, value, signal) => self.doStream(key, value, "parent", parentUpdater, signal),
stream: (key, value, signal) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.doStream(key, value, "self", parentUpdater, signal);
}
return self.doStream(key, value, "parent", parentUpdater, signal);
},
};
return parentUpdater;
@@ -94,34 +138,76 @@ export class StandardMetadataManager implements RunMetadataManager {
// Create the updater object and store it in a local variable
const rootUpdater: RunMetadataUpdater = {
set: (key, value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.set(key, value);
}
self.queuedRootOperations.add({ type: "set", key, value });
return rootUpdater;
},
del: (key) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.del(key);
}
self.queuedRootOperations.add({ type: "delete", key });
return rootUpdater;
},
append: (key, value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.append(key, value);
}
self.queuedRootOperations.add({ type: "append", key, value });
return rootUpdater;
},
remove: (key, value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.remove(key, value);
}
self.queuedRootOperations.add({ type: "remove", key, value });
return rootUpdater;
},
increment: (key, value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.increment(key, value);
}
self.queuedRootOperations.add({ type: "increment", key, value });
return rootUpdater;
},
decrement: (key, value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.decrement(key, value);
}
self.queuedRootOperations.add({ type: "increment", key, value: -Math.abs(value) });
return rootUpdater;
},
update: (value) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.update(value);
}
self.queuedRootOperations.add({ type: "update", value });
return rootUpdater;
},
stream: (key, value, signal) => self.doStream(key, value, "root", rootUpdater, signal),
stream: (key, value, signal) => {
// We have to check runIdIsRoot here because parent/root are executed before runIdIsRoot is set
if (self.runIdIsRoot) {
return self.doStream(key, value, "self", rootUpdater, signal);
}
return self.doStream(key, value, "root", rootUpdater, signal);
},
};
return rootUpdater;
packages/core/src/v3/schemas/common.ts
+132 -46
View File
@@ -219,50 +219,20 @@ export const TaskRun = z.object({
version: z.string().optional(),
metadata: z.record(DeserializedJsonSchema).optional(),
maxDuration: z.number().optional(),
/** @deprecated */
context: z.any(),
/**
* @deprecated For live values use the `usage` SDK functions
* @link https://trigger.dev/docs/run-usage
*/
durationMs: z.number().default(0),
/**
* @deprecated For live values use the `usage` SDK functions
* @link https://trigger.dev/docs/run-usage
*/
costInCents: z.number().default(0),
/**
* @deprecated For live values use the `usage` SDK functions
* @link https://trigger.dev/docs/run-usage
*/
baseCostInCents: z.number().default(0),
/** The priority of the run. Wih a value of 10 it will be dequeued before runs that were triggered 9 seconds before it (assuming they had no priority set). */
priority: z.number().optional(),
baseCostInCents: z.number().optional(),
parentTaskRunId: z.string().optional(),
rootTaskRunId: z.string().optional(),
// These are only used during execution, not in run.ctx
durationMs: z.number().optional(),
costInCents: z.number().optional(),
});
export type TaskRun = z.infer<typeof TaskRun>;
export const TaskRunExecutionTask = z.object({
id: z.string(),
filePath: z.string(),
exportName: z.string().optional(),
});
export type TaskRunExecutionTask = z.infer<typeof TaskRunExecutionTask>;
export const TaskRunExecutionAttempt = z.object({
number: z.number(),
startedAt: z.coerce.date(),
/** @deprecated */
id: z.string(),
/** @deprecated */
backgroundWorkerId: z.string(),
/** @deprecated */
backgroundWorkerTaskId: z.string(),
/** @deprecated */
status: z.string(),
});
export const GitMeta = z.object({
commitAuthorName: z.string().optional(),
commitMessage: z.string().optional(),
@@ -277,6 +247,18 @@ export const GitMeta = z.object({
export type GitMeta = z.infer<typeof GitMeta>;
export const TaskRunExecutionTask = z.object({
id: z.string(),
filePath: z.string(),
});
export type TaskRunExecutionTask = z.infer<typeof TaskRunExecutionTask>;
export const TaskRunExecutionAttempt = z.object({
number: z.number(),
startedAt: z.coerce.date(),
});
export type TaskRunExecutionAttempt = z.infer<typeof TaskRunExecutionAttempt>;
export const TaskRunExecutionEnvironment = z.object({
@@ -317,40 +299,144 @@ export const TaskRunExecutionBatch = z.object({
id: z.string(),
});
export const TaskRunExecution = z.object({
export const TaskRunExecutionDeployment = z.object({
id: z.string(),
shortCode: z.string(),
version: z.string(),
runtime: z.string(),
runtimeVersion: z.string(),
git: GitMeta.optional(),
});
export type TaskRunExecutionDeployment = z.infer<typeof TaskRunExecutionDeployment>;
const StaticTaskRunExecutionShape = {
task: TaskRunExecutionTask,
queue: TaskRunExecutionQueue,
environment: TaskRunExecutionEnvironment,
organization: TaskRunExecutionOrganization,
project: TaskRunExecutionProject,
machine: MachinePreset,
batch: TaskRunExecutionBatch.optional(),
deployment: TaskRunExecutionDeployment.optional(),
};
export const StaticTaskRunExecution = z.object(StaticTaskRunExecutionShape);
export type StaticTaskRunExecution = z.infer<typeof StaticTaskRunExecution>;
export const TaskRunExecution = z.object({
attempt: TaskRunExecutionAttempt,
run: TaskRun.and(
z.object({
traceContext: z.record(z.unknown()).optional(),
})
),
...StaticTaskRunExecutionShape,
});
export type TaskRunExecution = z.infer<typeof TaskRunExecution>;
export const V3TaskRunExecutionTask = z.object({
id: z.string(),
filePath: z.string(),
exportName: z.string().optional(),
});
export type V3TaskRunExecutionTask = z.infer<typeof V3TaskRunExecutionTask>;
export const V3TaskRunExecutionAttempt = z.object({
number: z.number(),
startedAt: z.coerce.date(),
id: z.string(),
backgroundWorkerId: z.string(),
backgroundWorkerTaskId: z.string(),
status: z.string(),
});
export type V3TaskRunExecutionAttempt = z.infer<typeof V3TaskRunExecutionAttempt>;
export const V3TaskRun = z.object({
id: z.string(),
payload: z.string(),
payloadType: z.string(),
tags: z.array(z.string()),
isTest: z.boolean().default(false),
createdAt: z.coerce.date(),
startedAt: z.coerce.date().default(() => new Date()),
idempotencyKey: z.string().optional(),
maxAttempts: z.number().optional(),
version: z.string().optional(),
metadata: z.record(DeserializedJsonSchema).optional(),
maxDuration: z.number().optional(),
context: z.unknown(),
durationMs: z.number(),
costInCents: z.number(),
baseCostInCents: z.number(),
});
export type V3TaskRun = z.infer<typeof V3TaskRun>;
export const V3TaskRunExecution = z.object({
task: V3TaskRunExecutionTask,
attempt: V3TaskRunExecutionAttempt,
run: V3TaskRun.and(
z.object({
traceContext: z.record(z.unknown()).optional(),
})
),
queue: TaskRunExecutionQueue,
environment: TaskRunExecutionEnvironment,
organization: TaskRunExecutionOrganization,
project: TaskRunExecutionProject,
batch: TaskRunExecutionBatch.optional(),
machine: MachinePreset,
batch: TaskRunExecutionBatch.optional(),
});
export type TaskRunExecution = z.infer<typeof TaskRunExecution>;
export type V3TaskRunExecution = z.infer<typeof V3TaskRunExecution>;
export const TaskRunContext = z.object({
task: TaskRunExecutionTask,
attempt: TaskRunExecutionAttempt.omit({
attempt: TaskRunExecutionAttempt,
run: TaskRun.omit({
payload: true,
payloadType: true,
metadata: true,
durationMs: true,
costInCents: true,
}),
...StaticTaskRunExecutionShape,
});
export type TaskRunContext = z.infer<typeof TaskRunContext>;
export const V3TaskRunExecutionEnvironment = z.object({
id: z.string(),
slug: z.string(),
type: z.enum(["PRODUCTION", "STAGING", "DEVELOPMENT", "PREVIEW"]),
});
export type V3TaskRunExecutionEnvironment = z.infer<typeof V3TaskRunExecutionEnvironment>;
export const V3TaskRunContext = z.object({
attempt: V3TaskRunExecutionAttempt.omit({
backgroundWorkerId: true,
backgroundWorkerTaskId: true,
}),
run: TaskRun.omit({ payload: true, payloadType: true, metadata: true }),
run: V3TaskRun.omit({
payload: true,
payloadType: true,
metadata: true,
}),
task: V3TaskRunExecutionTask,
queue: TaskRunExecutionQueue,
environment: TaskRunExecutionEnvironment,
environment: V3TaskRunExecutionEnvironment,
organization: TaskRunExecutionOrganization,
project: TaskRunExecutionProject,
batch: TaskRunExecutionBatch.optional(),
machine: MachinePreset.optional(),
});
export type TaskRunContext = z.infer<typeof TaskRunContext>;
export type V3TaskRunContext = z.infer<typeof V3TaskRunContext>;
export const TaskRunExecutionRetry = z.object({
timestamp: z.number(),
packages/core/src/v3/schemas/messages.ts
+11 -10
View File
@@ -6,12 +6,13 @@ import {
TaskRunExecutionResult,
TaskRunFailedExecutionResult,
TaskRunInternalError,
V3TaskRunExecution,
} from "./common.js";
import { TaskResource } from "./resources.js";
import {
EnvironmentType,
ProdTaskRunExecution,
ProdTaskRunExecutionPayload,
V3ProdTaskRunExecution,
V3ProdTaskRunExecutionPayload,
RunEngineVersionSchema,
TaskRunExecutionLazyAttemptPayload,
TaskRunExecutionMetrics,
@@ -83,7 +84,7 @@ export const BackgroundWorkerClientMessages = z.discriminatedUnion("type", [
version: z.literal("v1").default("v1"),
type: z.literal("TASK_RUN_COMPLETED"),
completion: TaskRunExecutionResult,
execution: TaskRunExecution,
execution: V3TaskRunExecution,
}),
z.object({
version: z.literal("v1").default("v1"),
@@ -368,7 +369,7 @@ export const CoordinatorToPlatformMessages = {
}),
z.object({
success: z.literal(true),
executionPayload: ProdTaskRunExecutionPayload,
executionPayload: V3ProdTaskRunExecutionPayload,
}),
]),
},
@@ -385,7 +386,7 @@ export const CoordinatorToPlatformMessages = {
}),
z.object({
success: z.literal(true),
payload: ProdTaskRunExecutionPayload,
payload: V3ProdTaskRunExecutionPayload,
}),
]),
},
@@ -417,7 +418,7 @@ export const CoordinatorToPlatformMessages = {
TASK_RUN_COMPLETED: {
message: z.object({
version: z.enum(["v1", "v2"]).default("v1"),
execution: ProdTaskRunExecution,
execution: V3ProdTaskRunExecution,
completion: TaskRunExecutionResult,
checkpoint: z
.object({
@@ -430,7 +431,7 @@ export const CoordinatorToPlatformMessages = {
TASK_RUN_COMPLETED_WITH_ACK: {
message: z.object({
version: z.enum(["v1", "v2"]).default("v2"),
execution: ProdTaskRunExecution,
execution: V3ProdTaskRunExecution,
completion: TaskRunExecutionResult,
checkpoint: z
.object({
@@ -720,7 +721,7 @@ export const ProdWorkerToCoordinatorMessages = {
TASK_RUN_COMPLETED: {
message: z.object({
version: z.enum(["v1", "v2"]).default("v1"),
execution: ProdTaskRunExecution,
execution: V3ProdTaskRunExecution,
completion: TaskRunExecutionResult,
}),
callback: z.object({
@@ -792,7 +793,7 @@ export const ProdWorkerToCoordinatorMessages = {
}),
z.object({
success: z.literal(true),
executionPayload: ProdTaskRunExecutionPayload,
executionPayload: V3ProdTaskRunExecutionPayload,
}),
]),
},
@@ -835,7 +836,7 @@ export const CoordinatorToProdWorkerMessages = {
EXECUTE_TASK_RUN: {
message: z.object({
version: z.literal("v1").default("v1"),
executionPayload: ProdTaskRunExecutionPayload,
executionPayload: V3ProdTaskRunExecutionPayload,
}),
},
EXECUTE_TASK_RUN_LAZY_ATTEMPT: {
packages/core/src/v3/schemas/schemas.ts
+12 -6
View File
@@ -1,6 +1,12 @@
import { z } from "zod";
import { RequireKeys } from "../types/index.js";
import { MachineConfig, MachinePreset, MachinePresetName, TaskRunExecution } from "./common.js";
import {
MachineConfig,
MachinePreset,
MachinePresetName,
TaskRunExecution,
V3TaskRunExecution,
} from "./common.js";
/*
WARNING: Never import anything from ./messages here. If it's needed in both, put it here instead.
@@ -36,7 +42,7 @@ export type TaskRunExecutionPayload = z.infer<typeof TaskRunExecutionPayload>;
// Strategies for not breaking backwards compatibility:
// 1. Add new fields as optional
// 2. If a field is required, add a default value
export const ProdTaskRunExecution = TaskRunExecution.extend({
export const V3ProdTaskRunExecution = V3TaskRunExecution.extend({
worker: z.object({
id: z.string(),
contentHash: z.string(),
@@ -46,16 +52,16 @@ export const ProdTaskRunExecution = TaskRunExecution.extend({
machine: MachinePreset.default({ name: "small-1x", cpu: 1, memory: 1, centsPerMs: 0 }),
});
export type ProdTaskRunExecution = z.infer<typeof ProdTaskRunExecution>;
export type V3ProdTaskRunExecution = z.infer<typeof V3ProdTaskRunExecution>;
export const ProdTaskRunExecutionPayload = z.object({
execution: ProdTaskRunExecution,
export const V3ProdTaskRunExecutionPayload = z.object({
execution: V3ProdTaskRunExecution,
traceContext: z.record(z.unknown()),
environment: z.record(z.string()).optional(),
metrics: TaskRunExecutionMetrics.optional(),
});
export type ProdTaskRunExecutionPayload = z.infer<typeof ProdTaskRunExecutionPayload>;
export type V3ProdTaskRunExecutionPayload = z.infer<typeof V3ProdTaskRunExecutionPayload>;
export const FixedWindowRateLimit = z.object({
type: z.literal("fixed-window"),
packages/core/src/v3/taskContext/index.ts
-2
View File
@@ -82,11 +82,9 @@ export class TaskContextAPI {
get contextAttributes(): Attributes {
if (this.ctx) {
return {
[SemanticInternalAttributes.ATTEMPT_ID]: this.ctx.attempt.id,
[SemanticInternalAttributes.ATTEMPT_NUMBER]: this.ctx.attempt.number,
[SemanticInternalAttributes.TASK_SLUG]: this.ctx.task.id,
[SemanticInternalAttributes.TASK_PATH]: this.ctx.task.filePath,
[SemanticInternalAttributes.TASK_EXPORT_NAME]: this.ctx.task.exportName,
[SemanticInternalAttributes.QUEUE_NAME]: this.ctx.queue.name,
[SemanticInternalAttributes.QUEUE_ID]: this.ctx.queue.id,
[SemanticInternalAttributes.RUN_ID]: this.ctx.run.id,
packages/core/src/v3/usage/api.ts
+5 -1
View File
@@ -1,7 +1,7 @@
const API_NAME = "usage";
import { getGlobal, registerGlobal, unregisterGlobal } from "../utils/globals.js";
import type { UsageManager, UsageMeasurement, UsageSample } from "./types.js";
import type { InitialUsageState, UsageManager, UsageMeasurement, UsageSample } from "./types.js";
import { NoopUsageManager } from "./noopUsageManager.js";
const NOOP_USAGE_MANAGER = new NoopUsageManager();
@@ -53,6 +53,10 @@ export class UsageAPI implements UsageManager {
this.disable();
}
public getInitialState(): InitialUsageState {
return this.#getUsageManager().getInitialState();
}
#getUsageManager(): UsageManager {
return getGlobal(API_NAME) ?? NOOP_USAGE_MANAGER;
}
packages/core/src/v3/usage/devUsageManager.ts
+17 -1
View File
@@ -1,4 +1,4 @@
import { UsageManager, UsageMeasurement, UsageSample } from "./types.js";
import { InitialUsageState, UsageManager, UsageMeasurement, UsageSample } from "./types.js";
import { clock } from "../clock-api.js";
import { ClockTime, calculateDurationInMs } from "../clock/clock.js";
@@ -45,6 +45,10 @@ export class DevUsageManager implements UsageManager {
private _firstMeasurement?: DevUsageMeasurement;
private _currentMeasurements: Map<string, DevUsageMeasurement> = new Map();
private _pauses: Map<string, { start: ClockTime; end?: ClockTime }> = new Map();
private _initialState: InitialUsageState = {
cpuTime: 0,
costInCents: 0,
};
disable(): void {}
@@ -54,6 +58,18 @@ export class DevUsageManager implements UsageManager {
this._firstMeasurement = undefined;
this._currentMeasurements.clear();
this._pauses.clear();
this._initialState = {
cpuTime: 0,
costInCents: 0,
};
}
setInitialState(state: InitialUsageState) {
this._initialState = state;
}
getInitialState(): InitialUsageState {
return this._initialState;
}
sample(): UsageSample | undefined {
packages/core/src/v3/usage/noopUsageManager.ts
+8 -1
View File
@@ -1,4 +1,4 @@
import { UsageManager, UsageMeasurement, UsageSample } from "./types.js";
import { InitialUsageState, UsageManager, UsageMeasurement, UsageSample } from "./types.js";
export class NoopUsageManager implements UsageManager {
disable(): void {
@@ -30,4 +30,11 @@ export class NoopUsageManager implements UsageManager {
reset(): void {
// Noop
}
getInitialState(): InitialUsageState {
return {
cpuTime: 0,
costInCents: 0,
};
}
}
packages/core/src/v3/usage/prodUsageManager.ts
+17 -1
View File
@@ -1,5 +1,5 @@
import { setInterval } from "node:timers/promises";
import { UsageManager, UsageMeasurement, UsageSample } from "./types.js";
import { InitialUsageState, UsageManager, UsageMeasurement, UsageSample } from "./types.js";
import { UsageClient } from "./usageClient.js";
export type ProdUsageManagerOptions = {
@@ -13,6 +13,10 @@ export class ProdUsageManager implements UsageManager {
private _abortController: AbortController | undefined;
private _lastSample: UsageSample | undefined;
private _usageClient: UsageClient | undefined;
private _initialState: InitialUsageState = {
cpuTime: 0,
costInCents: 0,
};
constructor(
private readonly delegageUsageManager: UsageManager,
@@ -27,6 +31,14 @@ export class ProdUsageManager implements UsageManager {
return typeof this._usageClient !== "undefined";
}
setInitialState(state: InitialUsageState) {
this._initialState = state;
}
getInitialState(): InitialUsageState {
return this._initialState;
}
reset(): void {
this.delegageUsageManager.reset();
this._abortController?.abort();
@@ -34,6 +46,10 @@ export class ProdUsageManager implements UsageManager {
this._usageClient = undefined;
this._measurement = undefined;
this._lastSample = undefined;
this._initialState = {
cpuTime: 0,
costInCents: 0,
};
}
disable(): void {
packages/core/src/v3/usage/types.ts
+6
View File
@@ -7,8 +7,14 @@ export interface UsageMeasurement {
sample(): UsageSample;
}
export type InitialUsageState = {
cpuTime: number;
costInCents: number;
};
export interface UsageManager {
disable(): void;
getInitialState(): InitialUsageState;
start(): UsageMeasurement;
stop(measurement: UsageMeasurement): UsageSample;
sample(): UsageSample | undefined;
packages/trigger-sdk/src/v3/shared.ts
-1
View File
@@ -1313,7 +1313,6 @@ async function triggerAndWait_internal<TIdentifier extends string, TPayload, TOu
{
payload: payloadPacket.data,
options: {
dependentAttempt: ctx.attempt.id,
lockToVersion: taskContext.worker?.version, // Lock to current version because we're waiting for it to finish
queue: options?.queue ? { name: options.queue } : undefined,
concurrencyKey: options?.concurrencyKey,
packages/trigger-sdk/src/v3/usage.ts
+7 -6
View File
@@ -54,6 +54,7 @@ export const usage = {
*/
getCurrent: (): CurrentUsage => {
const sample = usageApi.sample();
const initialState = usageApi.getInitialState();
const machine = taskContext.ctx?.machine;
const run = taskContext.ctx?.run;
@@ -65,12 +66,12 @@ export const usage = {
durationMs: 0,
},
total: {
costInCents: run?.costInCents ?? 0,
durationMs: run?.durationMs ?? 0,
costInCents: initialState.costInCents,
durationMs: initialState.cpuTime,
},
},
baseCostInCents: run?.baseCostInCents ?? 0,
totalCostInCents: (run?.costInCents ?? 0) + (run?.baseCostInCents ?? 0),
totalCostInCents: initialState.costInCents + (run?.baseCostInCents ?? 0),
};
}
@@ -83,12 +84,12 @@ export const usage = {
durationMs: sample.cpuTime,
},
total: {
costInCents: (run?.costInCents ?? 0) + currentCostInCents,
durationMs: (run?.durationMs ?? 0) + sample.cpuTime,
costInCents: currentCostInCents + initialState.costInCents,
durationMs: sample.cpuTime + initialState.cpuTime,
},
},
baseCostInCents: run?.baseCostInCents ?? 0,
totalCostInCents: (run?.costInCents ?? 0) + currentCostInCents + (run?.baseCostInCents ?? 0),
totalCostInCents: currentCostInCents + (run?.baseCostInCents ?? 0) + initialState.costInCents,
};
},
/**
pnpm-lock.yaml
Generated
+21 -3
View File
@@ -921,6 +921,24 @@ importers:
docs: {}
internal-packages/cache:
dependencies:
'@internal/redis':
specifier: workspace:*
version: link:../redis
'@trigger.dev/core':
specifier: workspace:*
version: link:../../packages/core
'@unkey/cache':
specifier: ^1.5.0
version: 1.5.0
'@unkey/error':
specifier: ^0.2.0
version: 0.2.0
superjson:
specifier: ^2.2.1
version: 2.2.1
internal-packages/clickhouse:
dependencies:
'@clickhouse/client':
@@ -1054,6 +1072,9 @@ importers:
internal-packages/run-engine:
dependencies:
'@internal/cache':
specifier: workspace:*
version: link:../cache
'@internal/redis':
specifier: workspace:*
version: link:../redis
@@ -1069,9 +1090,6 @@ importers:
'@trigger.dev/redis-worker':
specifier: workspace:*
version: link:../../packages/redis-worker
'@unkey/cache':
specifier: ^1.5.0
version: 1.5.0
assert-never:
specifier: ^1.2.1
version: 1.2.1
references/hello-world/src/trigger/example.ts
+2 -2
View File
@@ -57,7 +57,7 @@ export const parentTask = task({
id: "parent",
machine: "medium-1x",
run: async (payload: any, { ctx }) => {
logger.log("Hello, world from the parent", { payload });
logger.log("Hello, world from the parent", { payload, ctx });
await childTask.triggerAndWait({ message: "Hello, world!", aReallyBigInt: BigInt(10000) });
},
});
@@ -107,7 +107,7 @@ export const childTask = task({
}: { message?: string; failureChance?: number; duration?: number; aReallyBigInt?: bigint },
{ ctx }
) => {
logger.info("Hello, world from the child", { message, failureChance, aReallyBigInt });
logger.info("Hello, world from the child", { ctx, failureChance, aReallyBigInt });
if (Math.random() < failureChance) {
throw new Error("Random error at start");
references/hello-world/src/trigger/metadata.ts
+2
View File
@@ -44,6 +44,8 @@ export const parentTask = task({
metadata.parent.set("test.parent.set", true);
metadata.set("test.set", "test");
logger.info("logging metadata.current()", { current: metadata.current() });
await childTask.triggerAndWait({});
return {
references/hello-world/src/trigger/usage.ts
+35
View File
@@ -0,0 +1,35 @@
import { logger, task, wait, usage } from "@trigger.dev/sdk";
import { setTimeout } from "timers/promises";
export const usageExampleTask = task({
id: "usage-example",
retry: {
maxAttempts: 3,
minTimeoutInMs: 500,
maxTimeoutInMs: 1000,
factor: 1.5,
},
run: async (payload: { throwError: boolean }, { ctx }) => {
logger.info("run.ctx", { ctx });
await setTimeout(1000);
const currentUsage = usage.getCurrent();
logger.info("currentUsage", { currentUsage });
if (payload.throwError && ctx.attempt.number === 1) {
throw new Error("Forced error to cause a retry");
}
await setTimeout(5000);
const currentUsage2 = usage.getCurrent();
logger.info("currentUsage2", { currentUsage2 });
return {
message: "Hello, world!",
};
},
});