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

Adding missing task run hierarchy to TaskRun table (#1332)

Browse Source 
* Add task run hierarchical relationships to the database

* Add depth and related runs to the retrieve run API response

* Remove prisma optimize

* restructure the migrations to create the index concurrently

* Delete these tsbuildinfo files

* Fix type error by adding depth to the run list presenter

* Cleanup the task hierarchy, share more code

* Remove some fields from the list run response
This commit is contained in:
Eric Allam
2024-09-20 11:09:00 +01:00
committed by GitHub
parent 6976311e18
commit ba3c5bdf33
21 changed files with 518 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/components/runs/v3/RunInspector.tsx
+4
View File
@@ -354,6 +354,10 @@ export function RunInspector({
: "–"}
</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Run ID</Property.Label>
<Property.Value>{run.id}</Property.Value>
</Property.Item>
</Property.Table>
</div>
) : tab === "context" ? (
apps/webapp/app/db.server.ts
+21 -21
View File
@@ -1,10 +1,10 @@
import { PrismaClient, Prisma } from "@trigger.dev/database";
import { Prisma, PrismaClient } from "@trigger.dev/database";
import invariant from "tiny-invariant";
import { z } from "zod";
import { logger } from "./services/logger.server";
import { env } from "./env.server";
import { singleton } from "./utils/singleton";
import { logger } from "./services/logger.server";
import { isValidDatabaseUrl } from "./utils/db";
import { singleton } from "./utils/singleton";
export type PrismaTransactionClient = Omit<
PrismaClient,
@@ -94,6 +94,7 @@ function getClient() {
url: databaseUrl.href,
},
},
// @ts-expect-error
log: [
{
emit: "stdout",
@@ -107,25 +108,16 @@ function getClient() {
emit: "stdout",
level: "warn",
},
// {
// emit: "stdout",
// level: "query",
// },
// {
// emit: "event",
// level: "query",
// },
],
].concat(
process.env.VERBOSE_PRISMA_LOGS === "1"
? [
{ emit: "event", level: "query" },
{ emit: "stdout", level: "query" },
]
: []
),
});
// client.$on("query", (e) => {
// console.log(`Query tooks ${e.duration}ms`, {
// query: e.query,
// params: e.params,
// duration: e.duration,
// });
// });
// connect eagerly
client.$connect();
@@ -153,6 +145,7 @@ function getReplicaClient() {
url: replicaUrl.href,
},
},
// @ts-expect-error
log: [
{
emit: "stdout",
@@ -166,7 +159,14 @@ function getReplicaClient() {
emit: "stdout",
level: "warn",
},
],
].concat(
process.env.VERBOSE_PRISMA_LOGS === "1"
? [
{ emit: "event", level: "query" },
{ emit: "stdout", level: "query" },
]
: []
),
});
// connect eagerly
apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts
+110 -21
View File
@@ -4,6 +4,7 @@ import {
RunStatus,
SerializedError,
TaskRunError,
TriggerFunction,
conditionallyImportPacket,
createJsonErrorObject,
logger,
@@ -14,6 +15,47 @@ import assertNever from "assert-never";
import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
import { generatePresignedUrl } from "~/v3/r2.server";
import { BasePresenter } from "./basePresenter.server";
import { prisma } from "~/db.server";
// Build 'select' object
const commonRunSelect = {
id: true,
friendlyId: true,
status: true,
taskIdentifier: true,
createdAt: true,
startedAt: true,
updatedAt: true,
completedAt: true,
expiredAt: true,
delayUntil: true,
ttl: true,
tags: true,
costInCents: true,
baseCostInCents: true,
usageDurationMs: true,
idempotencyKey: true,
isTest: true,
depth: true,
lockedToVersion: {
select: {
version: true,
},
},
resumeParentOnCompletion: true,
batch: {
select: {
id: true,
friendlyId: true,
},
},
} satisfies Prisma.TaskRunSelect;
type CommonRelatedRun = Prisma.Result<
typeof prisma.taskRun,
{ select: typeof commonRunSelect },
"findFirstOrThrow"
>;
export class ApiRetrieveRunPresenter extends BasePresenter {
public async call(
@@ -22,7 +64,7 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
showSecretDetails: boolean
): Promise<RetrieveRunResponse | undefined> {
return this.traceWithEnv("call", env, async (span) => {
const taskRun = await this._prisma.taskRun.findUnique({
const taskRun = await this._replica.taskRun.findFirst({
where: {
friendlyId,
runtimeEnvironmentId: env.id,
@@ -36,6 +78,23 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
lockedToVersion: true,
schedule: true,
tags: true,
batch: {
select: {
id: true,
friendlyId: true,
},
},
parentTaskRun: {
select: commonRunSelect,
},
rootTaskRun: {
select: commonRunSelect,
},
childRuns: {
select: {
...commonRunSelect,
},
},
},
});
@@ -101,29 +160,11 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
const apiStatus = ApiRetrieveRunPresenter.apiStatusFromRunStatus(taskRun.status);
return {
id: taskRun.friendlyId,
status: apiStatus,
taskIdentifier: taskRun.taskIdentifier,
idempotencyKey: taskRun.idempotencyKey ?? undefined,
version: taskRun.lockedToVersion ? taskRun.lockedToVersion.version : undefined,
createdAt: taskRun.createdAt ?? undefined,
updatedAt: taskRun.updatedAt ?? undefined,
startedAt: taskRun.startedAt ?? taskRun.lockedAt ?? undefined,
finishedAt: ApiRetrieveRunPresenter.isStatusFinished(apiStatus)
? taskRun.updatedAt
: undefined,
delayedUntil: taskRun.delayUntil ?? undefined,
...createCommonRunStructure(taskRun),
payload: $payload,
payloadPresignedUrl: $payloadPresignedUrl,
output: $output,
outputPresignedUrl: $outputPresignedUrl,
isTest: taskRun.isTest,
ttl: taskRun.ttl ?? undefined,
expiredAt: taskRun.expiredAt ?? undefined,
tags: taskRun.tags.map((t) => t.name).sort((a, b) => a.localeCompare(b)),
costInCents: taskRun.costInCents,
baseCostInCents: taskRun.baseCostInCents,
durationMs: taskRun.usageDurationMs,
schedule: taskRun.schedule
? {
id: taskRun.schedule.friendlyId,
@@ -138,7 +179,6 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
},
}
: undefined,
...ApiRetrieveRunPresenter.apiBooleanHelpersFromRunStatus(apiStatus),
attempts: !showSecretDetails
? []
: taskRun.attempts.map((a) => ({
@@ -150,6 +190,13 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
completedAt: a.completedAt ?? undefined,
error: ApiRetrieveRunPresenter.apiErrorFromError(a.error),
})),
relatedRuns: {
root: taskRun.rootTaskRun ? createCommonRunStructure(taskRun.rootTaskRun) : undefined,
parent: taskRun.parentTaskRun
? createCommonRunStructure(taskRun.parentTaskRun)
: undefined,
children: taskRun.childRuns.map((r) => createCommonRunStructure(r)),
},
};
});
}
@@ -225,6 +272,12 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
}
}
static apiBooleanHelpersFromTaskRunStatus(status: TaskRunStatus) {
return ApiRetrieveRunPresenter.apiBooleanHelpersFromRunStatus(
ApiRetrieveRunPresenter.apiStatusFromRunStatus(status)
);
}
static apiBooleanHelpersFromRunStatus(status: RunStatus) {
const isQueued = status === "QUEUED" || status === "WAITING_FOR_DEPLOY" || status === "DELAYED";
const isExecuting = status === "EXECUTING" || status === "REATTEMPTING" || status === "FROZEN";
@@ -275,3 +328,39 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
}
}
}
function createCommonRunStructure(run: CommonRelatedRun) {
return {
id: run.friendlyId,
taskIdentifier: run.taskIdentifier,
idempotencyKey: run.idempotencyKey ?? undefined,
version: run.lockedToVersion?.version,
status: ApiRetrieveRunPresenter.apiStatusFromRunStatus(run.status),
createdAt: run.createdAt,
startedAt: run.startedAt ?? undefined,
updatedAt: run.updatedAt,
finishedAt: run.completedAt ?? undefined,
expiredAt: run.expiredAt ?? undefined,
delayedUntil: run.delayUntil ?? undefined,
ttl: run.ttl ?? undefined,
costInCents: run.costInCents,
baseCostInCents: run.baseCostInCents,
durationMs: run.usageDurationMs,
isTest: run.isTest,
depth: run.depth,
tags: run.tags
.map((t: { name: string }) => t.name)
.sort((a: string, b: string) => a.localeCompare(b)),
...ApiRetrieveRunPresenter.apiBooleanHelpersFromTaskRunStatus(run.status),
triggerFunction: resolveTriggerFunction(run),
batchId: run.batch?.friendlyId,
};
}
function resolveTriggerFunction(run: CommonRelatedRun): TriggerFunction {
if (run.batch) {
return run.resumeParentOnCompletion ? "batchTriggerAndWait" : "batchTrigger";
} else {
return run.resumeParentOnCompletion ? "triggerAndWait" : "trigger";
}
}
apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts
+1
View File
@@ -253,6 +253,7 @@ export class ApiRunListPresenter extends BasePresenter {
costInCents: run.costInCents,
baseCostInCents: run.baseCostInCents,
durationMs: run.usageDurationMs,
depth: run.depth,
...ApiRetrieveRunPresenter.apiBooleanHelpersFromRunStatus(
ApiRetrieveRunPresenter.apiStatusFromRunStatus(run.status)
),
apps/webapp/app/presenters/v3/RunListPresenter.server.ts
+3
View File
@@ -167,6 +167,7 @@ export class RunListPresenter extends BasePresenter {
baseCostInCents: number;
usageDurationMs: BigInt;
tags: string[];
depth: number;
}[]
>`
SELECT
@@ -190,6 +191,7 @@ export class RunListPresenter extends BasePresenter {
tr."baseCostInCents" AS "baseCostInCents",
tr."costInCents" AS "costInCents",
tr."usageDurationMs" AS "usageDurationMs",
tr."depth" AS "depth",
array_remove(array_agg(tag.name), NULL) AS "tags"
FROM
${sqlDatabaseSchema}."TaskRun" tr
@@ -333,6 +335,7 @@ WHERE
baseCostInCents: run.baseCostInCents,
usageDurationMs: Number(run.usageDurationMs),
tags: run.tags.sort((a, b) => a.localeCompare(b)),
depth: run.depth,
};
}),
pagination: {
apps/webapp/app/presenters/v3/SpanPresenter.server.ts
+1
View File
@@ -229,6 +229,7 @@ export class SpanPresenter extends BasePresenter {
};
return {
id: run.id,
friendlyId: run.friendlyId,
status: run.status,
createdAt: run.createdAt,
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.v3.$projectParam.runs.$runParam.spans.$spanParam/route.tsx
+8
View File
@@ -593,6 +593,14 @@ function RunBody({
: "–"}
</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Run ID</Property.Label>
<Property.Value>{run.friendlyId}</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Internal ID</Property.Label>
<Property.Value>{run.id}</Property.Value>
</Property.Item>
</Property.Table>
</div>
) : tab === "context" ? (
apps/webapp/app/v3/services/batchTriggerTask.server.ts
+1
View File
@@ -113,6 +113,7 @@ export class BatchTriggerTaskService extends BaseService {
options: {
...item.options,
dependentBatch: dependentAttempt?.id ? batch.friendlyId : undefined, // Only set dependentBatch if dependentAttempt is set which means batchTriggerAndWait was called
parentBatch: dependentAttempt?.id ? undefined : batch.friendlyId, // Only set parentBatch if dependentAttempt is NOT set which means batchTrigger was called
},
},
{
apps/webapp/app/v3/services/triggerTask.server.ts
+67
View File
@@ -108,6 +108,8 @@ export class TriggerTaskService extends BaseService {
id: true,
status: true,
taskIdentifier: true,
rootTaskRunId: true,
depth: true,
},
},
},
@@ -134,6 +136,23 @@ export class TriggerTaskService extends BaseService {
}
}
const parentAttempt = body.options?.parentAttempt
? await this._prisma.taskRunAttempt.findUnique({
where: { friendlyId: body.options.parentAttempt },
include: {
taskRun: {
select: {
id: true,
status: true,
taskIdentifier: true,
rootTaskRunId: true,
depth: true,
},
},
},
})
: undefined;
const dependentBatchRun = body.options?.dependentBatch
? await this._prisma.batchTaskRun.findUnique({
where: { friendlyId: body.options.dependentBatch },
@@ -145,6 +164,8 @@ export class TriggerTaskService extends BaseService {
id: true,
status: true,
taskIdentifier: true,
rootTaskRunId: true,
depth: true,
},
},
},
@@ -176,6 +197,26 @@ export class TriggerTaskService extends BaseService {
}
}
const parentBatchRun = body.options?.parentBatch
? await this._prisma.batchTaskRun.findUnique({
where: { friendlyId: body.options.parentBatch },
include: {
dependentTaskAttempt: {
include: {
taskRun: {
select: {
id: true,
status: true,
taskIdentifier: true,
rootTaskRunId: true,
},
},
},
},
},
})
: undefined;
return await eventRepository.traceEvent(
taskId,
{
@@ -243,6 +284,14 @@ export class TriggerTaskService extends BaseService {
}
}
const depth = dependentAttempt
? dependentAttempt.taskRun.depth + 1
: parentAttempt
? parentAttempt.taskRun.depth + 1
: dependentBatchRun?.dependentTaskAttempt
? dependentBatchRun.dependentTaskAttempt.taskRun.depth + 1
: 0;
const taskRun = await tx.taskRun.create({
data: {
status: delayUntil ? "DELAYED" : "PENDING",
@@ -272,6 +321,24 @@ export class TriggerTaskService extends BaseService {
: {
connect: tagIds.map((id) => ({ id })),
},
parentTaskRunId:
dependentAttempt?.taskRun.id ??
parentAttempt?.taskRun.id ??
dependentBatchRun?.dependentTaskAttempt?.taskRun.id,
parentTaskRunAttemptId:
dependentAttempt?.id ??
parentAttempt?.id ??
dependentBatchRun?.dependentTaskAttempt?.id,
rootTaskRunId:
dependentAttempt?.taskRun.rootTaskRunId ??
dependentAttempt?.taskRun.id ??
parentAttempt?.taskRun.rootTaskRunId ??
parentAttempt?.taskRun.id ??
dependentBatchRun?.dependentTaskAttempt?.taskRun.rootTaskRunId ??
dependentBatchRun?.dependentTaskAttempt?.taskRun.id,
batchId: dependentBatchRun?.id ?? parentBatchRun?.id,
resumeParentOnCompletion: !!(dependentAttempt ?? dependentBatchRun),
depth,
},
});
apps/webapp/package.json
+1 -1
View File
@@ -247,4 +247,4 @@
"engines": {
"node": ">=16.0.0"
}
}
}
docs/v3-openapi.yaml
+117 -80
View File
@@ -1821,7 +1821,7 @@ components:
format: date-time
description: The Date to delay the run until, e.g. `new Date()` or `"2024-06-25T15:45:26Z"`
example: 2024-06-25T15:45:26Z
RetrieveRunResponse:
CommonRunObject:
type: object
required:
- id
@@ -1829,7 +1829,6 @@ components:
- taskIdentifier
- createdAt
- updatedAt
- attempts
properties:
id:
type: string
@@ -1859,22 +1858,6 @@ components:
type: string
example: 20240523.1
description: The version of the worker that executed the run
payload:
type: object
description: The payload that was sent to the task. Will be omitted if the request was made with a Public API key
example: { "foo": "bar" }
payloadPresignedUrl:
type: string
description: The presigned URL to download the payload. Will only be included if the payload is too large to be included in the response. Expires in 5 minutes.
example: "https://r2.cloudflarestorage.com/packets/yubjwjsfkxnylobaqvqz/dev/run_p4omhh45hgxxnq1re6ovy/payload.json?X-Amz-Expires=300&X-Amz-Date=20240625T154526Z&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=10b064e58a0680db5b5e077be2be3b2a%2F20240625%2Fauto%2Fs3%2Faws4_request&X-Amz-SignedHeaders=host&X-Amz-Signature=88604cb993ffc151b4d73f2439da431d9928488e4b3dcfa4a7c8f1819"
output:
type: object
description: The output of the run. Will be omitted if the request was made with a Public API key
example: { "foo": "bar" }
outputPresignedUrl:
type: string
description: The presigned URL to download the output. Will only be included if the output is too large to be included in the response. Expires in 5 minutes.
example: "https://r2.cloudflarestorage.com/packets/yubjwjsfkxnylobaqvqz/dev/run_p4omhh45hgxxnq1re6ovy/payload.json?X-Amz-Expires=300&X-Amz-Date=20240625T154526Z&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=10b064e58a0680db5b5e077be2be3b2a%2F20240625%2Fauto%2Fs3%2Faws4_request&X-Amz-SignedHeaders=host&X-Amz-Signature=88604cb993ffc151b4d73f2439da431d9928488e4b3dcfa4a7c8f1819"
idempotencyKey:
type: string
description: The idempotency key used to prevent creating duplicate runs, if provided
@@ -1926,77 +1909,131 @@ components:
type: number
example: 491
description: The duration of compute (so far) in milliseconds. This does not include waits.
schedule:
type: object
description: The schedule that triggered the run. Will be omitted if the run was not triggered by a schedule
depth:
type: integer
example: 0
description: The depth of the run in the task run hierarchy. The root run has a depth of 0.
batchId:
type: string
description: The ID of the batch that this run belongs to
example: batch_1234
triggerFunction:
type: string
description: The name of the function that triggered the run
enum:
- trigger
- triggerAndWait
- batchTrigger
- batchTriggerAndWait
RetrieveRunResponse:
allOf:
- $ref: "#/components/schemas/CommonRunObject"
- type: object
required:
- id
- generator
- attempts
properties:
id:
payload:
type: object
description: The payload that was sent to the task. Will be omitted if the request was made with a Public API key
example: { "foo": "bar" }
payloadPresignedUrl:
type: string
description: The unique ID of the schedule, prefixed with `sched_`
example: sched_1234
externalId:
description: The presigned URL to download the payload. Will only be included if the payload is too large to be included in the response. Expires in 5 minutes.
example: "https://r2.cloudflarestorage.com/packets/yubjwjsfkxnylobaqvqz/dev/run_p4omhh45hgxxnq1re6ovy/payload.json?X-Amz-Expires=300&X-Amz-Date=20240625T154526Z&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=10b064e58a0680db5b5e077be2be3b2a%2F20240625%2Fauto%2Fs3%2Faws4_request&X-Amz-SignedHeaders=host&X-Amz-Signature=88604cb993ffc151b4d73f2439da431d9928488e4b3dcfa4a7c8f1819"
output:
type: object
description: The output of the run. Will be omitted if the request was made with a Public API key
example: { "foo": "bar" }
outputPresignedUrl:
type: string
description: The external ID of the schedule. Can be anything that is useful to you (e.g., user ID, org ID, etc.)
example: user_1234
deduplicationKey:
type: string
description: The deduplication key used to prevent creating duplicate schedules
example: dedup_key_1234
generator:
description: The presigned URL to download the output. Will only be included if the output is too large to be included in the response. Expires in 5 minutes.
example: "https://r2.cloudflarestorage.com/packets/yubjwjsfkxnylobaqvqz/dev/run_p4omhh45hgxxnq1re6ovy/payload.json?X-Amz-Expires=300&X-Amz-Date=20240625T154526Z&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=10b064e58a0680db5b5e077be2be3b2a%2F20240625%2Fauto%2Fs3%2Faws4_request&X-Amz-SignedHeaders=host&X-Amz-Signature=88604cb993ffc151b4d73f2439da431d9928488e4b3dcfa4a7c8f1819"
relatedRuns:
type: object
properties:
type:
root:
$ref: "#/components/schemas/CommonRunObject"
description: The root run of the run hierarchy. Will be omitted if the run is the root run
parent:
$ref: "#/components/schemas/CommonRunObject"
description: The parent run of the run. Will be omitted if the run is the root run
children:
description: The immediate children of the run. Will be omitted if the run has no children
type: array
items:
$ref: "#/components/schemas/CommonRunObject"
schedule:
type: object
description: The schedule that triggered the run. Will be omitted if the run was not triggered by a schedule
required:
- id
- generator
properties:
id:
type: string
enum:
- CRON
expression:
description: The unique ID of the schedule, prefixed with `sched_`
example: sched_1234
externalId:
type: string
description: The cron expression used to generate the schedule
example: 0 0 * * *
description:
description: The external ID of the schedule. Can be anything that is useful to you (e.g., user ID, org ID, etc.)
example: user_1234
deduplicationKey:
type: string
description: The description of the generator in plain english
example: Every day at midnight
attempts:
type: array
items:
type: object
required:
- id
- status
- createdAt
- updatedAt
properties:
id:
type: string
description: The unique ID of the attempt, prefixed with `attempt_`
example: attempt_1234
status:
type: string
enum:
- PENDING
- EXECUTING
- PAUSED
- COMPLETED
- FAILED
- CANCELED
error:
$ref: "#/components/schemas/SerializedError"
createdAt:
type: string
format: date-time
updatedAt:
type: string
format: date-time
startedAt:
type: string
format: date-time
completedAt:
type: string
format: date-time
description: The deduplication key used to prevent creating duplicate schedules
example: dedup_key_1234
generator:
type: object
properties:
type:
type: string
enum:
- CRON
expression:
type: string
description: The cron expression used to generate the schedule
example: 0 0 * * *
description:
type: string
description: The description of the generator in plain english
example: Every day at midnight
attempts:
type: array
items:
type: object
required:
- id
- status
- createdAt
- updatedAt
properties:
id:
type: string
description: The unique ID of the attempt, prefixed with `attempt_`
example: attempt_1234
status:
type: string
enum:
- PENDING
- EXECUTING
- PAUSED
- COMPLETED
- FAILED
- CANCELED
error:
$ref: "#/components/schemas/SerializedError"
createdAt:
type: string
format: date-time
updatedAt:
type: string
format: date-time
startedAt:
type: string
format: date-time
completedAt:
type: string
format: date-time
CreateScheduleOptions:
type: object
properties:
packages/build/tsconfig.src.tsbuildinfo
-1
View File
File diff suppressed because one or more lines are too long
packages/cli-v3/tsconfig.src.tsbuildinfo
-1
View File
File diff suppressed because one or more lines are too long
packages/core/src/v3/schemas/api.ts
+23
View File
@@ -68,7 +68,9 @@ export const TriggerTaskRequestBody = z.object({
options: z
.object({
dependentAttempt: z.string().optional(),
parentAttempt: z.string().optional(),
dependentBatch: z.string().optional(),
parentBatch: z.string().optional(),
lockToVersion: z.string().optional(),
queue: QueueOptions.optional(),
concurrencyKey: z.string().optional(),
@@ -470,6 +472,15 @@ export const RunScheduleDetails = z.object({
export type RunScheduleDetails = z.infer<typeof RunScheduleDetails>;
export const TriggerFunction = z.enum([
"triggerAndWait",
"trigger",
"batchTriggerAndWait",
"batchTrigger",
]);
export type TriggerFunction = z.infer<typeof TriggerFunction>;
const CommonRunFields = {
id: z.string(),
status: RunStatus,
@@ -496,6 +507,13 @@ const CommonRunFields = {
durationMs: z.number(),
};
export const RelatedRunDetails = z.object({
...CommonRunFields,
depth: z.number(),
triggerFunction: z.enum(["triggerAndWait", "trigger", "batchTriggerAndWait", "batchTrigger"]),
batchId: z.string().optional(),
});
export const RetrieveRunResponse = z.object({
...CommonRunFields,
payload: z.any().optional(),
@@ -503,6 +521,11 @@ export const RetrieveRunResponse = z.object({
output: z.any().optional(),
outputPresignedUrl: z.string().optional(),
schedule: RunScheduleDetails.optional(),
relatedRuns: z.object({
root: RelatedRunDetails.optional(),
parent: RelatedRunDetails.optional(),
children: z.array(RelatedRunDetails).optional(),
}),
attempts: z.array(
z
.object({
packages/core/tsconfig.src.tsbuildinfo
-1
View File
File diff suppressed because one or more lines are too long
packages/database/prisma/migrations/20240920085046_add_task_hierarchy_columns_without_parent_task_run_id_index/migration.sql
+19
View File
@@ -0,0 +1,19 @@
-- AlterTable
ALTER TABLE "TaskRun" ADD COLUMN "batchId" TEXT,
ADD COLUMN "depth" INTEGER NOT NULL DEFAULT 0,
ADD COLUMN "parentTaskRunAttemptId" TEXT,
ADD COLUMN "parentTaskRunId" TEXT,
ADD COLUMN "resumeParentOnCompletion" BOOLEAN NOT NULL DEFAULT false,
ADD COLUMN "rootTaskRunId" TEXT;
-- AddForeignKey
ALTER TABLE "TaskRun" ADD CONSTRAINT "TaskRun_rootTaskRunId_fkey" FOREIGN KEY ("rootTaskRunId") REFERENCES "TaskRun"("id") ON DELETE SET NULL ON UPDATE NO ACTION;
-- AddForeignKey
ALTER TABLE "TaskRun" ADD CONSTRAINT "TaskRun_parentTaskRunId_fkey" FOREIGN KEY ("parentTaskRunId") REFERENCES "TaskRun"("id") ON DELETE SET NULL ON UPDATE NO ACTION;
-- AddForeignKey
ALTER TABLE "TaskRun" ADD CONSTRAINT "TaskRun_parentTaskRunAttemptId_fkey" FOREIGN KEY ("parentTaskRunAttemptId") REFERENCES "TaskRunAttempt"("id") ON DELETE SET NULL ON UPDATE NO ACTION;
-- AddForeignKey
ALTER TABLE "TaskRun" ADD CONSTRAINT "TaskRun_batchId_fkey" FOREIGN KEY ("batchId") REFERENCES "BatchTaskRun"("id") ON DELETE SET NULL ON UPDATE NO ACTION;
packages/database/prisma/migrations/20240920085226_add_parent_task_run_id_index_concurrently/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- CreateIndex
CREATE INDEX CONCURRENTLY IF NOT EXISTS "TaskRun_parentTaskRunId_idx" ON "TaskRun"("parentTaskRunId");
packages/database/prisma/schema.prisma
+34 -2
View File
@@ -1720,7 +1720,37 @@ model TaskRun {
logsDeletedAt DateTime?
/// This represents the original task that that was triggered outside of a Trigger.dev task
rootTaskRun TaskRun? @relation("TaskRootRun", fields: [rootTaskRunId], references: [id], onDelete: SetNull, onUpdate: NoAction)
rootTaskRunId String?
/// The root run will have a list of all the descendant runs, children, grand children, etc.
descendantRuns TaskRun[] @relation("TaskRootRun")
/// The immediate parent run of this task run
parentTaskRun TaskRun? @relation("TaskParentRun", fields: [parentTaskRunId], references: [id], onDelete: SetNull, onUpdate: NoAction)
parentTaskRunId String?
/// The immediate child runs of this task run
childRuns TaskRun[] @relation("TaskParentRun")
/// The immediate parent attempt of this task run
parentTaskRunAttempt TaskRunAttempt? @relation("TaskParentRunAttempt", fields: [parentTaskRunAttemptId], references: [id], onDelete: SetNull, onUpdate: NoAction)
parentTaskRunAttemptId String?
/// The batch run that this task run is a part of
batch BatchTaskRun? @relation(fields: [batchId], references: [id], onDelete: SetNull, onUpdate: NoAction)
batchId String?
/// whether or not the task run was created because of a triggerAndWait for batchTriggerAndWait
resumeParentOnCompletion Boolean @default(false)
/// The depth of this task run in the task run hierarchy
depth Int @default(0)
@@unique([runtimeEnvironmentId, taskIdentifier, idempotencyKey])
// Finding child runs
@@index([parentTaskRunId])
// Task activity graph
@@index([projectId, createdAt, taskIdentifier])
//Runs list
@@ -1881,6 +1911,7 @@ model TaskRunAttempt {
batchTaskRunItems BatchTaskRunItem[]
CheckpointRestoreEvent CheckpointRestoreEvent[]
alerts ProjectAlert[]
childRuns TaskRun[] @relation("TaskParentRunAttempt")
@@unique([taskRunId, number])
@@index([taskRunId])
@@ -2071,8 +2102,9 @@ model BatchTaskRun {
items BatchTaskRunItem[]
runDependencies TaskRunDependency[] @relation("dependentBatchRun")
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
TaskRun TaskRun[]
@@unique([runtimeEnvironmentId, taskIdentifier, idempotencyKey])
}
packages/trigger-sdk/src/v3/shared.ts
+2
View File
@@ -791,6 +791,7 @@ async function trigger_internal<TPayload, TOutput>(
ttl: options?.ttl,
tags: options?.tags,
maxAttempts: options?.maxAttempts,
parentAttempt: taskContext.ctx?.attempt.id,
},
},
{
@@ -861,6 +862,7 @@ async function batchTrigger_internal<TPayload, TOutput>(
ttl: item.options?.ttl,
tags: item.options?.tags,
maxAttempts: item.options?.maxAttempts,
parentAttempt: taskContext.ctx?.attempt.id,
},
};
})
pnpm-lock.yaml
Generated
+3 -4
View File
@@ -5999,7 +5999,7 @@ packages:
engines: {node: ^12.22.0 || ^14.17.0 || >=16.0.0}
dependencies:
ajv: 6.12.6
debug: 4.3.6
debug: 4.3.7
espree: 9.6.0
globals: 13.19.0
ignore: 5.2.4
@@ -6230,7 +6230,7 @@ packages:
deprecated: Use @eslint/config-array instead
dependencies:
'@humanwhocodes/object-schema': 1.2.1
debug: 4.3.6
debug: 4.3.7
minimatch: 3.1.2
transitivePeerDependencies:
- supports-color
@@ -17679,7 +17679,6 @@ packages:
optional: true
dependencies:
ms: 2.1.3
dev: false
/decamelize-keys@1.1.1:
resolution: {integrity: sha512-WiPxgEirIV0/eIOMcnFBA3/IJZAZqKnwAwWyvvdi4lsr1WCN22nhdf/3db3DoZcUjTV2SqfzIwNyp6y2xs3nmg==}
@@ -19209,7 +19208,7 @@ packages:
ajv: 6.12.6
chalk: 4.1.2
cross-spawn: 7.0.3
debug: 4.3.6
debug: 4.3.7
doctrine: 3.0.0
escape-string-regexp: 4.0.0
eslint-scope: 7.2.0
references/v3-catalog/src/trigger/taskHierarchy.ts
+101
View File
@@ -0,0 +1,101 @@
import { runs, task } from "@trigger.dev/sdk/v3";
import { setTimeout } from "node:timers/promises";
export const rootTask = task({
id: "task-hierarchy/root-task",
run: async (
{ useWaits = true, useBatch = false }: { useWaits: boolean; useBatch: boolean },
{ ctx }
) => {
console.log("root-task");
if (useWaits) {
if (useBatch) {
await childTask.batchTriggerAndWait([{ payload: { useWaits, useBatch } }]);
} else {
await childTask.triggerAndWait({ useWaits, useBatch });
}
} else {
if (useBatch) {
await childTask.batchTrigger([{ payload: { useWaits, useBatch } }]);
} else {
await childTask.trigger({ useWaits, useBatch });
}
}
if (!useWaits) {
await setTimeout(10_000); // Wait for 10 seconds, all the runs will be finished by then
}
await logRunHierarchy(ctx.run.id);
},
});
export const childTask = task({
id: "task-hierarchy/child-task",
run: async ({ useWaits = true, useBatch = false }: { useWaits: boolean; useBatch: boolean }) => {
console.log("child-task");
if (useWaits) {
if (useBatch) {
await grandChildTask.batchTriggerAndWait([{ payload: { useWaits, useBatch } }]);
} else {
await grandChildTask.triggerAndWait({ useWaits, useBatch });
}
} else {
if (useBatch) {
await grandChildTask.batchTrigger([{ payload: { useWaits, useBatch } }]);
} else {
await grandChildTask.trigger({ useWaits, useBatch });
}
}
},
});
export const grandChildTask = task({
id: "task-hierarchy/grand-child-task",
run: async ({ useWaits = true, useBatch = false }: { useWaits: boolean; useBatch: boolean }) => {
console.log("grand-child-task");
if (useWaits) {
if (useBatch) {
await greatGrandChildTask.batchTriggerAndWait([{ payload: { useWaits, useBatch } }]);
} else {
await greatGrandChildTask.triggerAndWait({ useWaits, useBatch });
}
} else {
if (useBatch) {
await greatGrandChildTask.batchTrigger([{ payload: { useWaits, useBatch } }]);
} else {
await greatGrandChildTask.trigger({ useWaits, useBatch });
}
}
},
});
export const greatGrandChildTask = task({
id: "task-hierarchy/great-grand-child-task",
run: async ({ useWaits = true, useBatch = false }: { useWaits: boolean; useBatch: boolean }) => {
console.log("great-grand-child-task");
},
});
async function logRunHierarchy(
runId: string,
parentTaskIdentifier?: string,
triggerFunction?: string
) {
const runData = await runs.retrieve(runId);
const indent = " ".repeat(runData.depth * 2);
const triggerInfo = triggerFunction ? ` (triggered by ${triggerFunction})` : "";
const parentInfo = parentTaskIdentifier ? ` (parent task: ${parentTaskIdentifier})` : "";
console.log(
`${indent}Level ${runData.depth}: [${runData.taskIdentifier}] run ${runData.id}${triggerInfo}${parentInfo}`
);
for (const childRun of runData.relatedRuns.children ?? []) {
await logRunHierarchy(childRun.id, runData.taskIdentifier, childRun.triggerFunction);
}
}