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

v3: fix dependency trigger and wait (#1030)

Browse Source 
* fix dev dependecy resumes

* add catalog entry with multiple trigger waits

* update docker provider lifecycle command

* switch to x for clearing run filters

* Revert "fix dev dependecy resumes"

This reverts commit b4061f2ae71f21889adcf061928e2e77c436f0ed.

* fix dependency resumes, properly this time

* add catalog entry for dependency waits in loops

* advice in docs re parallel dependency waits

* fix link from v3 to v2 docs

* move lifecycle command logging to debug only

* Removed batchOptions from the trigger options and the docs

---------

Co-authored-by: Matt Aitken <matt@mattaitken.com>
This commit is contained in:
nicktrn
2024-04-18 14:48:43 +01:00
committed by GitHub
parent 2f5b4a8471
commit c9e1a3e9c5
12 changed files with 277 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/smart-needles-move.md
+5
View File
@@ -0,0 +1,5 @@
---
"@trigger.dev/sdk": patch
---
Remove unimplemented batchOptions
apps/docker-provider/src/index.ts
+1
View File
@@ -257,6 +257,7 @@ class DockerTaskOperations implements TaskOperations {
return await execa("docker", [
"exec",
containerName,
"busybox",
"wget",
"-q",
"-O-",
apps/kubernetes-provider/src/index.ts
+1 -1
View File
@@ -409,7 +409,7 @@ class KubernetesTaskOperations implements TaskOperations {
`for i in $(seq ${retries}); do sleep 1; busybox wget -q -O- 127.0.0.1:8000/${type}?cause=${cause} && break; done`,
];
logger.log("getLifecycleCommand()", { exec });
logger.debug("getLifecycleCommand()", { exec });
return exec;
}
apps/webapp/app/components/runs/v3/RunFilters.tsx
+2 -2
View File
@@ -1,4 +1,4 @@
import { TrashIcon } from "@heroicons/react/20/solid";
import { XMarkIcon } from "@heroicons/react/20/solid";
import { useNavigate } from "@remix-run/react";
import type { TaskRunStatus as TaskRunStatusType } from "@trigger.dev/database";
import { RuntimeEnvironment, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
@@ -247,7 +247,7 @@ export function RunsFilters({ possibleEnvironments, possibleTasks }: RunFiltersP
<TimeFrameFilter from={from} to={to} onRangeChanged={handleTimeFrameChange} />
<Button variant="minimal/small" onClick={() => clearFilters()} LeadingIcon={TrashIcon} />
<Button variant="minimal/small" onClick={() => clearFilters()} LeadingIcon={XMarkIcon} />
</div>
);
}
apps/webapp/app/v3/services/resumeAttempt.server.ts
+125 -122
View File
@@ -4,12 +4,13 @@ import {
TaskRunExecution,
TaskRunExecutionResult,
} from "@trigger.dev/core/v3";
import { $transaction } from "~/db.server";
import { $transaction, PrismaClientOrTransaction } from "~/db.server";
import { logger } from "~/services/logger.server";
import { marqs } from "~/v3/marqs/index.server";
import { socketIo } from "../handleSocketIo.server";
import { sharedQueueTasks } from "../marqs/sharedQueueConsumer.server";
import { BaseService } from "./baseService.server";
import { TaskRunAttempt } from "@trigger.dev/database";
export class ResumeAttemptService extends BaseService {
public async call(
@@ -24,7 +25,7 @@ export class ResumeAttemptService extends BaseService {
},
include: {
taskRun: true,
taskRunDependency: {
dependencies: {
include: {
taskRun: {
include: {
@@ -40,8 +41,12 @@ export class ResumeAttemptService extends BaseService {
},
},
},
orderBy: {
createdAt: "desc",
},
take: 1,
},
batchTaskRunDependency: {
batchDependencies: {
include: {
items: {
include: {
@@ -61,6 +66,10 @@ export class ResumeAttemptService extends BaseService {
},
},
},
orderBy: {
createdAt: "desc",
},
take: 1,
},
},
});
@@ -78,6 +87,8 @@ export class ResumeAttemptService extends BaseService {
return;
}
let completedAttemptIds: string[] = [];
switch (params.type) {
case "WAIT_FOR_DURATION": {
logger.error(
@@ -93,12 +104,10 @@ export class ResumeAttemptService extends BaseService {
});
break;
}
case "WAIT_FOR_TASK":
case "WAIT_FOR_BATCH": {
let completedAttemptIds: string[] = [];
if (attempt.taskRunDependency) {
const dependentAttempt = attempt.taskRunDependency.taskRun.attempts[0];
case "WAIT_FOR_TASK": {
if (attempt.dependencies.length) {
// We only care about the latest dependency
const dependentAttempt = attempt.dependencies[0].taskRun.attempts[0];
if (!dependentAttempt) {
logger.error("No dependent attempt", { attemptId: attempt.id });
@@ -106,19 +115,16 @@ export class ResumeAttemptService extends BaseService {
}
completedAttemptIds = [dependentAttempt.id];
await tx.taskRunAttempt.update({
where: {
id: attempt.id,
},
data: {
taskRunDependency: {
disconnect: true,
},
},
});
} else if (attempt.batchTaskRunDependency) {
const dependentBatchItems = attempt.batchTaskRunDependency.items;
} else {
logger.error("No task dependency", { attemptId: attempt.id });
return;
}
break;
}
case "WAIT_FOR_BATCH": {
if (attempt.batchDependencies) {
// We only care about the latest batch dependency
const dependentBatchItems = attempt.batchDependencies[0].items;
if (!dependentBatchItems) {
logger.error("No dependent batch items", { attemptId: attempt.id });
@@ -126,115 +132,112 @@ export class ResumeAttemptService extends BaseService {
}
completedAttemptIds = dependentBatchItems.map((item) => item.taskRun.attempts[0]?.id);
await tx.taskRunAttempt.update({
where: {
id: attempt.id,
},
data: {
batchTaskRunDependency: {
disconnect: true,
},
},
});
} else {
logger.error("No dependencies", { attemptId: attempt.id });
logger.error("No batch dependency", { attemptId: attempt.id });
return;
}
if (completedAttemptIds.length === 0) {
logger.error("No completed attempt IDs", { attemptId: attempt.id });
return;
}
const completions: TaskRunExecutionResult[] = [];
const executions: TaskRunExecution[] = [];
for (const completedAttemptId of completedAttemptIds) {
const completedAttempt = await tx.taskRunAttempt.findUnique({
where: {
id: completedAttemptId,
taskRun: {
lockedAt: {
not: null,
},
lockedById: {
not: null,
},
},
},
});
if (!completedAttempt) {
logger.error("Completed attempt not found", {
attemptId: attempt.id,
completedAttemptId,
});
await marqs?.acknowledgeMessage(attempt.taskRunId);
return;
}
const completion = await sharedQueueTasks.getCompletionPayloadFromAttempt(
completedAttempt.id
);
if (!completion) {
logger.error("Failed to get completion payload", {
attemptId: attempt.id,
completedAttemptId,
});
await marqs?.acknowledgeMessage(attempt.taskRunId);
return;
}
completions.push(completion);
const executionPayload = await sharedQueueTasks.getExecutionPayloadFromAttempt(
completedAttempt.id
);
if (!executionPayload) {
logger.error("Failed to get execution payload", {
attemptId: attempt.id,
completedAttemptId,
});
await marqs?.acknowledgeMessage(attempt.taskRunId);
return;
}
executions.push(executionPayload.execution);
}
const updated = await tx.taskRunAttempt.update({
where: {
id: attempt.id,
},
data: {
status: "EXECUTING",
taskRun: {
update: {
data: {
status: attempt.number > 1 ? "RETRYING_AFTER_FAILURE" : "EXECUTING",
},
},
},
},
});
socketIo.coordinatorNamespace.emit("RESUME_AFTER_DEPENDENCY", {
version: "v1",
runId: attempt.taskRunId,
attemptId: attempt.id,
attemptFriendlyId: attempt.friendlyId,
completions,
executions,
});
break;
}
default: {
break;
}
}
await this.#handleDependencyResume(attempt, completedAttemptIds, tx);
});
}
async #handleDependencyResume(
attempt: TaskRunAttempt,
completedAttemptIds: string[],
tx: PrismaClientOrTransaction
) {
if (completedAttemptIds.length === 0) {
logger.error("No completed attempt IDs", { attemptId: attempt.id });
return;
}
const completions: TaskRunExecutionResult[] = [];
const executions: TaskRunExecution[] = [];
for (const completedAttemptId of completedAttemptIds) {
const completedAttempt = await tx.taskRunAttempt.findUnique({
where: {
id: completedAttemptId,
taskRun: {
lockedAt: {
not: null,
},
lockedById: {
not: null,
},
},
},
});
if (!completedAttempt) {
logger.error("Completed attempt not found", {
attemptId: attempt.id,
completedAttemptId,
});
await marqs?.acknowledgeMessage(attempt.taskRunId);
return;
}
const completion = await sharedQueueTasks.getCompletionPayloadFromAttempt(
completedAttempt.id
);
if (!completion) {
logger.error("Failed to get completion payload", {
attemptId: attempt.id,
completedAttemptId,
});
await marqs?.acknowledgeMessage(attempt.taskRunId);
return;
}
completions.push(completion);
const executionPayload = await sharedQueueTasks.getExecutionPayloadFromAttempt(
completedAttempt.id
);
if (!executionPayload) {
logger.error("Failed to get execution payload", {
attemptId: attempt.id,
completedAttemptId,
});
await marqs?.acknowledgeMessage(attempt.taskRunId);
return;
}
executions.push(executionPayload.execution);
}
const updated = await tx.taskRunAttempt.update({
where: {
id: attempt.id,
},
data: {
status: "EXECUTING",
taskRun: {
update: {
data: {
status: attempt.number > 1 ? "RETRYING_AFTER_FAILURE" : "EXECUTING",
},
},
},
},
});
socketIo.coordinatorNamespace.emit("RESUME_AFTER_DEPENDENCY", {
version: "v1",
runId: attempt.taskRunId,
attemptId: attempt.id,
attemptFriendlyId: attempt.friendlyId,
completions,
executions,
});
}
}
docs/mint.json
+1 -1
View File
@@ -49,7 +49,7 @@
},
{
"name": "v2",
"url": "https://trigger.dev/docs",
"url": "https://trigger.dev/docs/documentation",
"version": "v3 (Developer Preview)"
},
{
docs/v3/triggering.mdx
+80
View File
@@ -165,6 +165,45 @@ export const myTask = task({
This is where it gets interesting. You can trigger a task and then wait for the result. This is useful when you need to call a different task and then use the result to continue with your task.
<Accordion title="Don't use this in parallel, e.g. with `Promise.all()`">
Instead, use `batchTriggerAndWait()` if you can, or a for loop if you can't.
To control concurrency using batch triggers, you can set `queue.concurrencyLimit` on the child task.
<CodeGroup>
```ts /trigger/batch.ts
export const batchTask = task({
id: "batch-task",
run: async (payload: string) => {
const results = await childTask.batchTriggerAndWait({
items: [{ payload: "item1" }, { payload: "item2" }],
});
console.log("Results", results);
//...do stuff with the results
},
});
```
```ts /trigger/loop.ts
export const loopTask = task({
id: "loop-task",
run: async (payload: string) => {
//this will be slower than the batch version
//as we have to resume the parent after each iteration
for (let i = 0; i < 2; i++) {
const result = await childTask.triggerAndWait({ payload: `item${i}` });
console.log("Result", result);
//...do stuff with the result
}
},
});
```
</CodeGroup>
</Accordion>
```ts /trigger/parent.ts
export const parentTask = task({
id: "parent-task",
@@ -181,6 +220,47 @@ export const parentTask = task({
You can batch trigger a task and wait for all the results. This is useful for the fan-out pattern, where you need to call a task multiple times and then wait for all the results to continue with your task.
<Accordion title="Don't use this in parallel, e.g. with `Promise.all()`">
Instead, pass in all items at once and set an appropriate `maxConcurrency`. Alternatively, use sequentially with a for loop.
To control concurrency, you can set `queue.concurrencyLimit` on the child task.
<CodeGroup>
```ts /trigger/batch.ts
export const batchTask = task({
id: "batch-task",
run: async (payload: string) => {
const results = await childTask.batchTriggerAndWait({
items: [{ payload: "item1" }, { payload: "item2" }],
});
console.log("Results", results);
//...do stuff with the results
},
});
```
```ts /trigger/loop.ts
export const loopTask = task({
id: "loop-task",
run: async (payload: string) => {
//this will be slower than a single batchTriggerAndWait()
//as we have to resume the parent after each iteration
for (let i = 0; i < 2; i++) {
const result = await childTask.batchTriggerAndWait({
items: [{ payload: `itemA${i}` }, { payload: `itemB${i}` }],
});
console.log("Result", result);
//...do stuff with the result
}
},
});
```
</CodeGroup>
</Accordion>
```ts /trigger/nested.ts
export const batchParentTask = task({
id: "parent-task",
packages/database/prisma/migrations/20240415135132_make_attempt_dependencies_an_array/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- DropIndex
DROP INDEX "TaskRunDependency_dependentAttemptId_key";
packages/database/prisma/migrations/20240415143325_remove_attempt_unique_constraint_from_batch_runs/migration.sql
+2
View File
@@ -0,0 +1,2 @@
-- DropIndex
DROP INDEX "BatchTaskRun_dependentTaskAttemptId_key";
packages/database/prisma/schema.prisma
+5 -5
View File
@@ -1686,8 +1686,8 @@ model TaskRunDependency {
checkpointEventId String? @unique
/// An attempt that is dependent on this task run.
dependentAttempt TaskRunAttempt? @relation("dependentAttempt", fields: [dependentAttemptId], references: [id])
dependentAttemptId String? @unique
dependentAttempt TaskRunAttempt? @relation(fields: [dependentAttemptId], references: [id])
dependentAttemptId String?
/// A batch run that is dependent on this task run
dependentBatchRun BatchTaskRun? @relation("dependentBatchRun", fields: [dependentBatchRunId], references: [id])
@@ -1751,8 +1751,8 @@ model TaskRunAttempt {
output String?
outputType String @default("application/json")
taskRunDependency TaskRunDependency? @relation("dependentAttempt")
batchTaskRunDependency BatchTaskRun?
dependencies TaskRunDependency[]
batchDependencies BatchTaskRun[]
checkpoints Checkpoint[]
batchTaskRunItems BatchTaskRunItem[]
@@ -1923,7 +1923,7 @@ model BatchTaskRun {
runtimeEnvironmentId String
dependentTaskAttempt TaskRunAttempt? @relation(fields: [dependentTaskAttemptId], references: [id], onDelete: Cascade, onUpdate: Cascade)
dependentTaskAttemptId String? @unique
dependentTaskAttemptId String?
items BatchTaskRunItem[]
runDependencies TaskRunDependency[] @relation("dependentBatchRun")
packages/trigger-sdk/src/v3/shared.ts
+2 -2
View File
@@ -179,12 +179,12 @@ export interface Task<TInput, TOutput = any> {
trigger: (params: { payload: TInput; options?: TaskRunOptions }) => Promise<InvokeHandle>;
batchTrigger: (params: {
items: { payload: TInput; options?: TaskRunOptions }[];
batchOptions?: BatchRunOptions;
// batchOptions?: BatchRunOptions;
}) => Promise<InvokeBatchHandle>;
triggerAndWait: (params: { payload: TInput; options?: TaskRunOptions }) => Promise<TOutput>;
batchTriggerAndWait: (params: {
items: { payload: TInput; options?: TaskRunOptions }[];
batchOptions?: BatchRunOptions;
// batchOptions?: BatchRunOptions;
}) => Promise<BatchResult<TOutput>>;
}
references/v3-catalog/src/trigger/subtasks.ts
+51
View File
@@ -114,3 +114,54 @@ export const subtasksWithRetries = task({
};
},
});
export const multipleTriggerWaits = task({
id: "multiple-trigger-waits",
run: async ({ message = "test" }: { message?: string }) => {
await simpleChildTask.triggerAndWait({ payload: { message: `${message} - 1.a` } });
await simpleChildTask.triggerAndWait({ payload: { message: `${message} - 2.a` } });
await simpleChildTask.batchTriggerAndWait({
items: [
{ payload: { message: `${message} - 3.a` } },
{ payload: { message: `${message} - 3.b` } },
],
});
await simpleChildTask.batchTriggerAndWait({
items: [
{ payload: { message: `${message} - 4.a` } },
{ payload: { message: `${message} - 4.b` } },
],
});
return {
hello: "world",
};
},
});
export const triggerAndWaitLoops = task({
id: "trigger-wait-loops",
run: async ({ message = "test" }: { message?: string }) => {
for (let i = 0; i < 2; i++) {
await simpleChildTask.triggerAndWait({ payload: { message: `${message} - ${i}` } });
}
for (let i = 0; i < 2; i++) {
await simpleChildTask.batchTriggerAndWait({
items: [
{ payload: { message: `${message} - ${i}.a` } },
{ payload: { message: `${message} - ${i}.b` } },
],
// batchOptions: { maxConcurrency: 1 },
});
}
// Don't do this!
// await Promise.all(
// [{ message: `${message} - 1` }, { message: `${message} - 2` }].map((payload) =>
// simpleChildTask.triggerAndWait({ payload })
// )
// );
},
});