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

Updates the trigger, batchTrigger and their *AndWait variants to use the first parameter for the payload/items, and the second parameter for options (#1045)

Browse Source 
Also always returns a `TaskRunResult` object from `triggerAndWait` instead of rethrowing subtask errors in the parent
This commit is contained in:
Eric Allam
2024-04-19 14:51:51 +01:00
committed by GitHub
parent b82db67b81
commit 374edef020
18 changed files with 377 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/shaggy-spoons-taste.md
+56
View File
@@ -0,0 +1,56 @@
---
"@trigger.dev/sdk": patch
"@trigger.dev/core": patch
---
Updates the `trigger`, `batchTrigger` and their `*AndWait` variants to use the first parameter for the payload/items, and the second parameter for options.
Before:
```ts
await yourTask.trigger({ payload: { foo: "bar" }, options: { idempotencyKey: "key_1234" } });
await yourTask.triggerAndWait({ payload: { foo: "bar" }, options: { idempotencyKey: "key_1234" } });
await yourTask.batchTrigger({ items: [{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }] });
await yourTask.batchTriggerAndWait({ items: [{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }] });
```
After:
```ts
await yourTask.trigger({ foo: "bar" }, { idempotencyKey: "key_1234" });
await yourTask.triggerAndWait({ foo: "bar" }, { idempotencyKey: "key_1234" });
await yourTask.batchTrigger([{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }]);
await yourTask.batchTriggerAndWait([{ payload: { foo: "bar" } }, { payload: { foo: "baz" } }]);
```
We've also changed the API of the `triggerAndWait` result. Before, if the subtask that was triggered finished with an error, we would automatically "rethrow" the error in the parent task.
Now instead we're returning a `TaskRunResult` object that allows you to discriminate between successful and failed runs in the subtask:
Before:
```ts
try {
const result = await yourTask.triggerAndWait({ foo: "bar" });
// result is the output of your task
console.log("result", result);
} catch (error) {
// handle subtask errors here
}
```
After:
```ts
const result = await yourTask.triggerAndWait({ foo: "bar" });
if (result.ok) {
console.log(`Run ${result.id} succeeded with output`, result.output);
} else {
console.log(`Run ${result.id} failed with error`, result.error);
}
```
docs/v3/errors-retrying.mdx
+1 -1
View File
@@ -35,7 +35,7 @@ export const myTask = task({
maxAttempts: 10,
},
run: async (payload: string) => {
const result = await otherTask.triggerAndWait({ payload: "some data" });
const result = await otherTask.triggerAndWait("some data");
//...do other stuff
},
});
docs/v3/migration-defer.mdx
+3 -3
View File
@@ -71,7 +71,7 @@ export async function runLongRunningTask() {
}
```
In Trigger.dev your logic goes in the `run` function of a task. You can then `trigger` and `batchTrigger` that task, with a payload and options.
In Trigger.dev your logic goes in the `run` function of a task. You can then `trigger` and `batchTrigger` that task, with a payload as the first argument.
```ts /app/actions/actions.ts
"use server";
@@ -79,7 +79,7 @@ In Trigger.dev your logic goes in the `run` function of a task. You can then `tr
import { longRunningTask } from "@/trigger/someTasks";
export async function runLongRunningTask() {
return await longRunningTask.trigger({ payload: { foo: "bar" } });
return await longRunningTask.trigger({ foo: "bar" });
}
```
@@ -243,7 +243,7 @@ export const longRunningTask = task({
import { longRunningTask } from "@/trigger/longRunningTask";
export async function runLongRunningTask() {
return await longRunningTask.trigger({ payload: { foo: "bar" } });
return await longRunningTask.trigger({ foo: "bar" });
}
```
docs/v3/queue-concurrency.mdx
+19 -30
View File
@@ -107,24 +107,19 @@ export async function POST(request: Request) {
if (data.branch === "main") {
//trigger the task, with a different queue
const handle = await generatePullRequest.trigger({
payload: data,
options: {
queue: {
//the "main-branch" queue will have a concurrency limit of 10
//this triggered run will use that queue
name: "main-branch",
concurrencyLimit: 10,
},
const handle = await generatePullRequest.trigger(data, {
queue: {
//the "main-branch" queue will have a concurrency limit of 10
//this triggered run will use that queue
name: "main-branch",
concurrencyLimit: 10,
},
});
return Response.json(handle);
} else {
//triggered with the default (concurrency of 1)
const handle = await generatePullRequest.trigger({
payload: data,
});
const handle = await generatePullRequest.trigger(data);
return Response.json(handle);
}
}
@@ -146,32 +141,26 @@ export async function POST(request: Request) {
if (data.isFreeUser) {
//free users can only have 1 PR generated at a time
const handle = await generatePullRequest.trigger({
payload: data,
options: {
queue: {
//every free user gets a queue with a concurrency limit of 1
name: "free-users",
concurrencyLimit: 1,
},
concurrencyKey: data.userId,
const handle = await generatePullRequest.trigger(data, {
queue: {
//every free user gets a queue with a concurrency limit of 1
name: "free-users",
concurrencyLimit: 1,
},
concurrencyKey: data.userId,
});
//return a success response with the handle
return Response.json(handle);
} else {
//trigger the task, with a different queue
const handle = await generatePullRequest.trigger({
payload: data,
options: {
queue: {
//every paid user gets a queue with a concurrency limit of 10
name: "paid-users",
concurrencyLimit: 10,
},
concurrencyKey: data.userId,
const handle = await generatePullRequest.trigger(data, {
queue: {
//every paid user gets a queue with a concurrency limit of 10
name: "paid-users",
concurrencyLimit: 10,
},
concurrencyKey: data.userId,
});
//return a success response with the handle
docs/v3/tasks-overview.mdx
+1 -1
View File
@@ -37,7 +37,7 @@ import { helloWorldTask } from "./trigger/hello-world";
async function triggerHelloWorld() {
//This triggers the task and return a handle
const handle = await helloWorld.trigger({ payload: { message: "Hello world!" } });
const handle = await helloWorld.trigger({ message: "Hello world!" });
//You can use the handle to check the status of the task, cancel and retry it.
console.log("Task is running with handle", handle.id);
docs/v3/triggering.mdx
+36 -29
View File
@@ -48,7 +48,7 @@ export async function POST(request: Request) {
const data = await request.json();
//trigger your task
const handle = await emailSequence.trigger({ payload: { to: data.email, name: data.name } });
const handle = await emailSequence.trigger({ to: data.email, name: data.name });
//return a success response with the handle
return Response.json(handle);
@@ -67,7 +67,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
const data = await request.json();
//trigger your task
const handle = await emailSequence.trigger({ payload: { to: data.email, name: data.name } });
const handle = await emailSequence.trigger({ to: data.email, name: data.name });
//return a success response with the handle
return json(handle);
@@ -91,9 +91,9 @@ export async function POST(request: Request) {
const data = await request.json();
//batch trigger your task
const batchHandle = await emailSequence.batchTrigger({
items: data.users.map((u) => ({ payload: { to: u.email, name: u.name } })),
});
const batchHandle = await emailSequence.batchTrigger(
data.users.map((u) => ({ payload: { to: u.email, name: u.name } }))
);
//return a success response with the handle
return Response.json(batchHandle);
@@ -112,9 +112,9 @@ export async function action({ request, params }: ActionFunctionArgs) {
const data = await request.json();
//batch trigger your task
const batchHandle = await emailSequence.batchTrigger({
items: data.users.map((u) => ({ payload: { to: u.email, name: u.name } })),
});
const batchHandle = await emailSequence.batchTrigger(
data.users.map((u) => ({ payload: { to: u.email, name: u.name } }))
);
//return a success response with the handle
return json(batchHandle);
@@ -137,7 +137,7 @@ import { myOtherTask } from "~/trigger/my-other-task";
export const myTask = task({
id: "my-task",
run: async (payload: string) => {
const handle = await myOtherTask.trigger({ payload: "some data" });
const handle = await myOtherTask.trigger("some data");
//...do other stuff
},
@@ -154,7 +154,7 @@ import { myOtherTask } from "~/trigger/my-other-task";
export const myTask = task({
id: "my-task",
run: async (payload: string) => {
const batchHandle = await myOtherTask.batchTrigger({ items: [{ payload: "some data" }] });
const batchHandle = await myOtherTask.batchTrigger([{ payload: "some data" }]);
//...do other stuff
},
@@ -168,16 +168,18 @@ This is where it gets interesting. You can trigger a task and then wait for the
<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.
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" }],
});
const results = await childTask.batchTriggerAndWait([
{ payload: "item1" },
{ payload: "item2" },
]);
console.log("Results", results);
//...do stuff with the results
@@ -192,7 +194,7 @@ export const loopTask = task({
//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}` });
const result = await childTask.triggerAndWait(`item${i}`);
console.log("Result", result);
//...do stuff with the result
@@ -200,6 +202,7 @@ export const loopTask = task({
},
});
```
</CodeGroup>
</Accordion>
@@ -208,7 +211,7 @@ export const loopTask = task({
export const parentTask = task({
id: "parent-task",
run: async (payload: string) => {
const result = await batchChildTask.triggerAndWait({ payload: "some-data" });
const result = await batchChildTask.triggerAndWait("some-data");
console.log("Result", result);
//...do stuff with the result
@@ -223,16 +226,18 @@ You can batch trigger a task and wait for all the results. This is useful for th
<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.
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" }],
});
const results = await childTask.batchTriggerAndWait([
{ payload: "item1" },
{ payload: "item2" },
]);
console.log("Results", results);
//...do stuff with the results
@@ -247,9 +252,10 @@ export const loopTask = task({
//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}` }],
});
const result = await childTask.batchTriggerAndWait([
{ payload: `itemA${i}` },
{ payload: `itemB${i}` },
]);
console.log("Result", result);
//...do stuff with the result
@@ -257,6 +263,7 @@ export const loopTask = task({
},
});
```
</CodeGroup>
</Accordion>
@@ -265,9 +272,11 @@ export const loopTask = task({
export const batchParentTask = task({
id: "parent-task",
run: async (payload: string) => {
const results = await childTask.batchTriggerAndWait({
items: [{ payload: "item4" }, { payload: "item5" }, { payload: "item6" }],
});
const results = await childTask.batchTriggerAndWait([
{ payload: "item4" },
{ payload: "item5" },
{ payload: "item6" },
]);
console.log("Results", results);
//...do stuff with the result
@@ -326,9 +335,7 @@ import { createAvatar } from "@/trigger/create-avatar";
export async function create() {
try {
const handle = await createAvatar.trigger({
payload: {
userImage: "http://...",
},
userImage: "http://...",
});
return { handle };
docs/v3/upgrading-from-v2.mdx
+1 -3
View File
@@ -164,9 +164,7 @@ We've unified triggering in v3. You use `trigger()` or `batchTrigger()` which yo
async function yourBackendFunction() {
//call `trigger()` on any task
const handle = await openaiTask.trigger({
payload: {
prompt: "Tell me a programming joke",
},
prompt: "Tell me a programming joke",
});
}
```
packages/core/src/v3/runtime/devRuntimeManager.ts
+4 -15
View File
@@ -8,10 +8,7 @@ import { RuntimeManager } from "./manager";
import { unboundedTimeout } from "../utils/timers";
export class DevRuntimeManager implements RuntimeManager {
_taskWaits: Map<
string,
{ resolve: (value: TaskRunExecutionResult) => void; reject?: (err?: any) => void }
> = new Map();
_taskWaits: Map<string, { resolve: (value: TaskRunExecutionResult) => void }> = new Map();
_batchWaits: Map<
string,
@@ -41,8 +38,8 @@ export class DevRuntimeManager implements RuntimeManager {
return pendingCompletion;
}
const promise = new Promise<TaskRunExecutionResult>((resolve, reject) => {
this._taskWaits.set(params.id, { resolve, reject });
const promise = new Promise<TaskRunExecutionResult>((resolve) => {
this._taskWaits.set(params.id, { resolve });
});
return await promise;
@@ -93,15 +90,7 @@ export class DevRuntimeManager implements RuntimeManager {
return;
}
if (!wait.reject) {
wait.resolve(completion);
} else {
if (completion.ok) {
wait.resolve(completion);
} else {
wait.reject(completion);
}
}
wait.resolve(completion);
this._taskWaits.delete(execution.run.id);
}
packages/core/src/v3/runtime/prodRuntimeManager.ts
+5 -17
View File
@@ -1,4 +1,3 @@
import { setTimeout } from "node:timers/promises";
import { clock } from "../clock-api";
import {
BatchTaskRunExecutionResult,
@@ -8,19 +7,16 @@ import {
TaskRunExecution,
TaskRunExecutionResult,
} from "../schemas";
import { unboundedTimeout } from "../utils/timers";
import { ZodIpcConnection } from "../zodIpc";
import { RuntimeManager } from "./manager";
import { unboundedTimeout } from "../utils/timers";
export type ProdRuntimeManagerOptions = {
waitThresholdInMs?: number;
};
export class ProdRuntimeManager implements RuntimeManager {
_taskWaits: Map<
string,
{ resolve: (value: TaskRunExecutionResult) => void; reject?: (err?: any) => void }
> = new Map();
_taskWaits: Map<string, { resolve: (value: TaskRunExecutionResult) => void }> = new Map();
_batchWaits: Map<
string,
@@ -91,8 +87,8 @@ export class ProdRuntimeManager implements RuntimeManager {
}
async waitForTask(params: { id: string; ctx: TaskRunContext }): Promise<TaskRunExecutionResult> {
const promise = new Promise<TaskRunExecutionResult>((resolve, reject) => {
this._taskWaits.set(params.id, { resolve, reject });
const promise = new Promise<TaskRunExecutionResult>((resolve) => {
this._taskWaits.set(params.id, { resolve });
});
await this.ipc.send("WAIT_FOR_TASK", {
@@ -139,15 +135,7 @@ export class ProdRuntimeManager implements RuntimeManager {
return;
}
if (!wait.reject) {
wait.resolve(completion);
} else {
if (completion.ok) {
wait.resolve(completion);
} else {
wait.reject(completion);
}
}
wait.resolve(completion);
this._taskWaits.delete(execution.run.id);
}
packages/trigger-sdk/src/v3/shared.ts
+78 -43
View File
@@ -49,7 +49,11 @@ export function queue(options: { name: string } & QueueOptions): Queue {
return options;
}
export type TaskOptions<TPayload, TOutput = any, TInitOutput extends InitOutput = any> = {
export type TaskOptions<
TPayload = void,
TOutput = unknown,
TInitOutput extends InitOutput = any,
> = {
/** An id for your task. This must be unique inside your project and not change between versions. */
id: string;
/** The retry settings when an uncaught error is thrown.
@@ -168,7 +172,7 @@ export type TaskRunResult<TOutput = any> =
| {
ok: false;
id: string;
error: any;
error: unknown;
};
export type BatchResult<TOutput = any> = {
@@ -176,18 +180,72 @@ export type BatchResult<TOutput = any> = {
runs: TaskRunResult<TOutput>[];
};
export interface Task<TInput, TOutput = any> {
type BatchItem<TInput> = TInput extends void
? { payload?: TInput; options?: TaskRunOptions }
: { payload: TInput; options?: TaskRunOptions };
export interface Task<TInput = void, TOutput = any> {
/**
* The id of the task.
*/
id: string;
trigger: (params: { payload: TInput; options?: TaskRunOptions }) => Promise<InvokeHandle>;
batchTrigger: (params: {
items: { payload: TInput; options?: TaskRunOptions }[];
// batchOptions?: BatchRunOptions;
}) => Promise<InvokeBatchHandle>;
triggerAndWait: (params: { payload: TInput; options?: TaskRunOptions }) => Promise<TOutput>;
batchTriggerAndWait: (params: {
items: { payload: TInput; options?: TaskRunOptions }[];
// batchOptions?: BatchRunOptions;
}) => Promise<BatchResult<TOutput>>;
/**
* Trigger a task with the given payload, and continue without waiting for the result. If you want to wait for the result, use `triggerAndWait`. Returns the id of the triggered task run.
* @param payload
* @param options
* @returns InvokeHandle
* - `id` - The id of the triggered task run.
*/
trigger: (payload: TInput, options?: TaskRunOptions) => Promise<InvokeHandle>;
/**
* Batch trigger multiple task runs with the given payloads, and continue without waiting for the results. If you want to wait for the results, use `batchTriggerAndWait`. Returns the id of the triggered batch.
* @param items
* @returns InvokeBatchHandle
* - `batchId` - The id of the triggered batch.
* - `runs` - The ids of the triggered task runs.
*/
batchTrigger: (items: Array<BatchItem<TInput>>) => Promise<InvokeBatchHandle>;
/**
* Trigger a task with the given payload, and wait for the result. Returns the result of the task run
* @param payload
* @param options - Options for the task run
* @returns TaskRunResult
* @example
* ```
* const result = await task.triggerAndWait({ foo: "bar" });
*
* if (result.ok) {
* console.log(result.output);
* } else {
* console.error(result.error);
* }
* ```
*/
triggerAndWait: (payload: TInput, options?: TaskRunOptions) => Promise<TaskRunResult<TOutput>>;
/**
* Batch trigger multiple task runs with the given payloads, and wait for the results. Returns the results of the task runs.
* @param items
* @returns BatchResult
* @example
* ```
* const result = await task.batchTriggerAndWait([
* { payload: { foo: "bar" } },
* { payload: { foo: "baz" } },
* ]);
*
* for (const run of result.runs) {
* if (run.ok) {
* console.log(run.output);
* } else {
* console.error(run.error);
* }
* }
* ```
*/
batchTriggerAndWait: (items: Array<BatchItem<TInput>>) => Promise<BatchResult<TOutput>>;
}
type TaskRunOptions = {
@@ -201,10 +259,6 @@ type TaskRunOptions = {
type TaskRunConcurrencyOptions = Queue;
type BatchRunOptions = TaskRunOptions & {
maxConcurrency?: number;
};
export type Prettify<T> = {
[K in keyof T]: T[K];
} & {};
@@ -213,12 +267,12 @@ export type DynamicBaseOptions = {
id: string;
};
export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
export function createTask<TInput = void, TOutput = unknown, TInitOutput extends InitOutput = any>(
params: TaskOptions<TInput, TOutput, TInitOutput>
): Task<TInput, TOutput> {
const task: Task<TInput, TOutput> = {
id: params.id,
trigger: async ({ payload, options }) => {
trigger: async (payload, options) => {
const apiClient = apiClientManager.client;
if (!apiClient) {
@@ -258,7 +312,6 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
[SemanticInternalAttributes.STYLE_ICON]: "trigger",
["messaging.client_id"]: taskContextManager.worker?.id,
[SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id,
["messaging.message.body.size"]: JSON.stringify(payload).length,
[SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
...(taskMetadata
? accessoryAttributes({
@@ -277,7 +330,7 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
return handle;
},
batchTrigger: async ({ items }) => {
batchTrigger: async (items) => {
const apiClient = apiClientManager.client;
if (!apiClient) {
@@ -323,9 +376,6 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
["messaging.batch.message_count"]: items.length,
["messaging.client_id"]: taskContextManager.worker?.id,
[SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id,
["messaging.message.body.size"]: items
.map((item) => JSON.stringify(item.payload))
.join("").length,
[SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
[SemanticInternalAttributes.STYLE_ICON]: "trigger",
...(taskMetadata
@@ -345,7 +395,7 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
return response;
},
triggerAndWait: async ({ payload, options }) => {
triggerAndWait: async (payload, options) => {
const ctx = taskContextManager.ctx;
if (!ctx) {
@@ -393,13 +443,7 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
}
);
const runResult = await handleTaskRunExecutionResult<TOutput>(result);
if (!runResult.ok) {
throw runResult.error;
}
return runResult.output;
return await handleTaskRunExecutionResult<TOutput>(result);
}
}
@@ -408,13 +452,7 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
ctx,
});
const runResult = await handleTaskRunExecutionResult<TOutput>(result);
if (!runResult.ok) {
throw runResult.error;
}
return runResult.output;
return await handleTaskRunExecutionResult<TOutput>(result);
},
{
kind: SpanKind.PRODUCER,
@@ -439,7 +477,7 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
}
);
},
batchTriggerAndWait: async ({ items }) => {
batchTriggerAndWait: async (items) => {
const ctx = taskContextManager.ctx;
if (!ctx) {
@@ -555,9 +593,6 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
["messaging.batch.message_count"]: items.length,
["messaging.client_id"]: taskContextManager.worker?.id,
[SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id,
["messaging.message.body.size"]: items
.map((item) => JSON.stringify(item.payload))
.join("").length,
[SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
[SemanticInternalAttributes.STYLE_ICON]: "trigger",
...(taskMetadata
packages/trigger-sdk/src/v3/tasks.ts
+1 -1
View File
@@ -19,7 +19,7 @@ import { TaskOptions, Task, createTask } from "./shared";
*
* @returns A task that can be triggered
*/
export function task<TInput, TOutput = any, TInitOutput extends InitOutput = any>(
export function task<TInput = void, TOutput = unknown, TInitOutput extends InitOutput = any>(
options: TaskOptions<TInput, TOutput, TInitOutput>
): Task<TInput, TOutput> {
return createTask<TInput, TOutput, TInitOutput>(options);
references/v3-catalog/src/trigger/batch.ts
+10 -6
View File
@@ -3,18 +3,22 @@ import { logger, task, wait } from "@trigger.dev/sdk/v3";
export const batchParentTask = task({
id: "batch-parent-task",
run: async () => {
const response = await batchChildTask.batchTrigger({
items: [{ payload: "item1" }, { payload: "item2" }, { payload: "item3" }],
});
const response = await batchChildTask.batchTrigger([
{ payload: "item1" },
{ payload: "item2" },
{ payload: "item3" },
]);
logger.info("Batch task response", { response });
await wait.for({ seconds: 5 });
await wait.until({ date: new Date(Date.now() + 1000 * 5) }); // 5 seconds
const waitResponse = await batchChildTask.batchTriggerAndWait({
items: [{ payload: "item4" }, { payload: "item5" }, { payload: "item6" }],
});
const waitResponse = await batchChildTask.batchTriggerAndWait([
{ payload: "item4" },
{ payload: "item5" },
{ payload: "item6" },
]);
logger.info("Batch task wait response", { waitResponse });
references/v3-catalog/src/trigger/concurrency.ts
+4 -4
View File
@@ -23,13 +23,13 @@ export const testConcurrency = task({
await new Promise((resolve) => setTimeout(resolve, 3000));
await testConcurrencyChild.batchTrigger({
items: Array.from({ length: count }).map((_, index) => ({
await testConcurrencyChild.batchTrigger(
Array.from({ length: count }).map((_, index) => ({
payload: {
delay,
},
})),
});
}))
);
logger.info(`All ${count} tasks triggered`);
references/v3-catalog/src/trigger/idempotencyKeys.ts
+16 -10
View File
@@ -1,19 +1,25 @@
import { task, wait } from "@trigger.dev/sdk/v3";
import { logger, task, wait } from "@trigger.dev/sdk/v3";
export const idempotencyKeyParent = task({
id: "idempotency-key-parent",
run: async (payload: { key: string }) => {
console.log("Hello from idempotency-key-parent");
const childTaskResponse = await idempotencyKeyChild.triggerAndWait({
payload: {
const childTaskResponse = await idempotencyKeyChild.triggerAndWait(
{
key: payload.key,
forceError: true,
},
options: {
{
idempotencyKey: payload.key,
},
});
}
);
if (childTaskResponse.ok) {
logger.log("Child task response", { output: childTaskResponse.output });
} else {
logger.error("Child task error", { error: childTaskResponse.error });
}
return {
key: payload.key,
@@ -42,8 +48,8 @@ export const idempotencyKeyBatchParent = task({
run: async (payload: { keyPrefix: string; itemCount: number }) => {
console.log("Hello from idempotency-key-batch-parent");
const childTaskResponse = await idempotencyKeyBatchChild.batchTriggerAndWait({
items: Array.from({ length: payload.itemCount }).map((_, index) => ({
const childTaskResponse = await idempotencyKeyBatchChild.batchTriggerAndWait(
Array.from({ length: payload.itemCount }).map((_, index) => ({
payload: {
key: `${payload.keyPrefix}-${index}`,
forceError: index % 2 === 0,
@@ -52,8 +58,8 @@ export const idempotencyKeyBatchParent = task({
options: {
idempotencyKey: `${payload.keyPrefix}-${index}`,
},
})),
});
}))
);
return {
keyPrefix: payload.keyPrefix,
references/v3-catalog/src/trigger/longRunning.ts
+1 -1
View File
@@ -19,7 +19,7 @@ export const longRunningParent = task({
run: async (payload: { message: string }) => {
logger.info("Long running parent", { payload });
await longRunning.triggerAndWait({ payload: { message: "child" } });
await longRunning.triggerAndWait({ message: "child" });
return {
finished: new Date().toISOString(),
references/v3-catalog/src/trigger/simple.ts
+4 -8
View File
@@ -83,19 +83,15 @@ export const parentTask = task({
await wait.for({ seconds: 5 });
const childTaskResponse = await childTask.triggerAndWait({
payload: {
message: payload.message,
forceError: false,
},
message: payload.message,
forceError: false,
});
logger.info("Child task response", { childTaskResponse });
await childTask.trigger({
payload: {
message: `${payload.message} - 2.a`,
forceError: true,
},
message: `${payload.message} - 2.a`,
forceError: true,
});
return {
references/v3-catalog/src/trigger/subtasks.ts
+80 -91
View File
@@ -5,36 +5,28 @@ export const simpleParentTask = task({
id: "simple-parent-task",
run: async (payload: { message: string }) => {
await simpleChildTask.trigger({
payload: {
message: `${payload.message} - 2.a`,
},
message: `${payload.message} - 2.a`,
});
await simpleChildTask.triggerAndWait({
payload: {
message: `${payload.message} - 2.b`,
message: `${payload.message} - 2.b`,
});
await simpleChildTask.batchTrigger([
{
payload: {
message: `${payload.message} - 2.c`,
},
},
});
]);
await simpleChildTask.batchTrigger({
items: [
{
payload: {
message: `${payload.message} - 2.c`,
},
await simpleChildTask.batchTriggerAndWait([
{
payload: {
message: `${payload.message} - 2.d`,
},
],
});
await simpleChildTask.batchTriggerAndWait({
items: [
{
payload: {
message: `${payload.message} - 2.d`,
},
},
],
});
},
]);
return {
hello: "world",
@@ -53,61 +45,51 @@ export const subtasksWithRetries = task({
id: "subtasks-with-retries",
run: async (payload: { message: string }) => {
await taskWithRetries.triggerAndWait({
payload: {
message: `${payload.message} - 2.b`,
message: `${payload.message} - 2.b`,
});
await taskWithRetries.batchTrigger([
{
payload: {
message: `${payload.message} - 2.c`,
},
},
});
{
payload: {
message: `${payload.message} - 2.cc`,
},
},
]);
await taskWithRetries.batchTrigger({
items: [
{
payload: {
message: `${payload.message} - 2.c`,
},
await taskWithRetries.batchTriggerAndWait([
{
payload: {
message: `${payload.message} - 2.d`,
},
{
payload: {
message: `${payload.message} - 2.cc`,
},
},
{
payload: {
message: `${payload.message} - 2.dd`,
},
],
});
await taskWithRetries.batchTriggerAndWait({
items: [
{
payload: {
message: `${payload.message} - 2.d`,
},
},
{
payload: {
message: `${payload.message} - 2.dd`,
},
},
],
});
},
]);
await taskWithRetries.triggerAndWait({
payload: {
message: `${payload.message} - 2.e`,
},
message: `${payload.message} - 2.e`,
});
await taskWithRetries.batchTriggerAndWait({
items: [
{
payload: {
message: `${payload.message} - 2.f`,
},
await taskWithRetries.batchTriggerAndWait([
{
payload: {
message: `${payload.message} - 2.f`,
},
{
payload: {
message: `${payload.message} - 2.ff`,
},
},
{
payload: {
message: `${payload.message} - 2.ff`,
},
],
});
},
]);
return {
hello: "world",
@@ -118,21 +100,17 @@ 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.triggerAndWait({ message: `${message} - 1.a` });
await simpleChildTask.triggerAndWait({ 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` } },
],
});
await simpleChildTask.batchTriggerAndWait([
{ payload: { message: `${message} - 3.a` } },
{ payload: { message: `${message} - 3.b` } },
]);
await simpleChildTask.batchTriggerAndWait([
{ payload: { message: `${message} - 4.a` } },
{ payload: { message: `${message} - 4.b` } },
]);
return {
hello: "world",
@@ -144,19 +122,21 @@ 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}` } });
await simpleChildTask.triggerAndWait({ 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 },
});
await simpleChildTask.batchTriggerAndWait([
{ payload: { message: `${message} - ${i}.a` } },
{ payload: { message: `${message} - ${i}.b` } },
]);
}
await taskWithNoPayload.trigger();
await taskWithNoPayload.triggerAndWait();
await taskWithNoPayload.batchTrigger([{}]);
await taskWithNoPayload.batchTriggerAndWait([{}]);
// Don't do this!
// await Promise.all(
// [{ message: `${message} - 1` }, { message: `${message} - 2` }].map((payload) =>
@@ -165,3 +145,12 @@ export const triggerAndWaitLoops = task({
// );
},
});
export const taskWithNoPayload = task({
id: "task-with-no-payload",
run: async () => {
logger.log("Task with no payload");
return { hello: "world" };
},
});
references/v3-catalog/src/trigger/superjson.ts
+57 -63
View File
@@ -4,19 +4,19 @@ export const superParentTask = task({
id: "super-parent-task",
run: async () => {
const result = await superChildTask.triggerAndWait({
payload: {
foo: "bar",
whenToDo: new Date(),
},
foo: "bar",
whenToDo: new Date(),
});
logger.log(`typeof result.date = ${typeof result.date}`);
logger.log(`typeof result.regex = ${typeof result.regex}`);
logger.log(`typeof result.bigint = ${typeof result.bigint}`);
logger.log(`typeof result.set = ${typeof result.set}`);
logger.log(`typeof result.map = ${typeof result.map}`);
logger.log(`typeof result.error = ${typeof result.error}`);
logger.log(`typeof result.url = ${typeof result.url}`);
if (result.ok) {
logger.log(`typeof result.date = ${typeof result.output.date}`);
logger.log(`typeof result.output.regex = ${typeof result.output.regex}`);
logger.log(`typeof result.output.bigint = ${typeof result.output.bigint}`);
logger.log(`typeof result.output.set = ${typeof result.output.set}`);
logger.log(`typeof result.output.map = ${typeof result.output.map}`);
logger.log(`typeof result.output.error = ${typeof result.output.error}`);
logger.log(`typeof result.output.url = ${typeof result.output.url}`);
}
return "## super-parent-task completed";
},
@@ -49,64 +49,60 @@ export const superHugePayloadTask = task({
run: async () => {
const largePayload = createLargeObject(1000, 128);
const result = await superHugeOutputTask.triggerAndWait({
payload: largePayload,
});
const result = await superHugeOutputTask.triggerAndWait(largePayload);
logger.log("Result from superHugeOutputTask: ", { result });
const batchResult = await superHugeOutputTask.batchTriggerAndWait({
items: [
{ payload: largePayload },
{
payload: {
small: "object",
},
const batchResult = await superHugeOutputTask.batchTriggerAndWait([
{ payload: largePayload },
{
payload: {
small: "object",
},
{ payload: largePayload },
{
payload: {
small: "object",
},
},
{ payload: largePayload },
{
payload: {
small: "object",
},
{ payload: largePayload },
{
payload: {
small: "object",
},
},
{ payload: largePayload },
{
payload: {
small: "object",
},
{ payload: largePayload },
{
payload: {
small: "object",
},
},
{ payload: largePayload },
{
payload: {
small: "object",
},
{ payload: largePayload },
{
payload: {
small: "object",
},
},
{ payload: largePayload },
{
payload: {
small: "object",
},
{ payload: largePayload },
{
payload: {
small: "object",
},
},
{ payload: largePayload },
{
payload: {
small: "object",
},
{ payload: largePayload },
{
payload: {
small: "object",
},
},
{ payload: largePayload },
{
payload: {
small: "object",
},
{ payload: largePayload },
{
payload: {
small: "object",
},
},
{ payload: largePayload },
{
payload: {
small: "object",
},
],
});
},
]);
logger.log("Result from superHugeOutputTask batchTriggerAndWait: ", { batchResult });
@@ -118,7 +114,7 @@ export const superHugePayloadTask = task({
export const superHugeOutputTask = task({
id: "super-huge-output-task",
run: async (payload) => {
run: async (payload: any) => {
return payload;
},
});
@@ -127,9 +123,7 @@ export const superStringTask = task({
id: "super-string-parent-task",
run: async () => {
const result = await superStringChildTask.triggerAndWait({
payload: {
foo: "bar",
},
foo: "bar",
});
return result;
@@ -138,7 +132,7 @@ export const superStringTask = task({
export const superStringChildTask = task({
id: "super-string-child-task",
run: async () => {
run: async (payload: any) => {
return "## super-string-child-task completed";
},
});