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, maxAttempts: 10,
}, },
run: async (payload: string) => { run: async (payload: string) => {
const result = await otherTask.triggerAndWait({ payload: "some data" }); const result = await otherTask.triggerAndWait("some data");
//...do other stuff //...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 ```ts /app/actions/actions.ts
"use server"; "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"; import { longRunningTask } from "@/trigger/someTasks";
export async function runLongRunningTask() { 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"; import { longRunningTask } from "@/trigger/longRunningTask";
export async function runLongRunningTask() { export async function runLongRunningTask() {
return await longRunningTask.trigger({ payload: { foo: "bar" } }); return await longRunningTask.trigger({ foo: "bar" });
} }
``` ```
docs/v3/queue-concurrency.mdx
+4 -15
View File
@@ -107,24 +107,19 @@ export async function POST(request: Request) {
if (data.branch === "main") { if (data.branch === "main") {
//trigger the task, with a different queue //trigger the task, with a different queue
const handle = await generatePullRequest.trigger({ const handle = await generatePullRequest.trigger(data, {
payload: data,
options: {
queue: { queue: {
//the "main-branch" queue will have a concurrency limit of 10 //the "main-branch" queue will have a concurrency limit of 10
//this triggered run will use that queue //this triggered run will use that queue
name: "main-branch", name: "main-branch",
concurrencyLimit: 10, concurrencyLimit: 10,
}, },
},
}); });
return Response.json(handle); return Response.json(handle);
} else { } else {
//triggered with the default (concurrency of 1) //triggered with the default (concurrency of 1)
const handle = await generatePullRequest.trigger({ const handle = await generatePullRequest.trigger(data);
payload: data,
});
return Response.json(handle); return Response.json(handle);
} }
} }
@@ -146,32 +141,26 @@ export async function POST(request: Request) {
if (data.isFreeUser) { if (data.isFreeUser) {
//free users can only have 1 PR generated at a time //free users can only have 1 PR generated at a time
const handle = await generatePullRequest.trigger({ const handle = await generatePullRequest.trigger(data, {
payload: data,
options: {
queue: { queue: {
//every free user gets a queue with a concurrency limit of 1 //every free user gets a queue with a concurrency limit of 1
name: "free-users", name: "free-users",
concurrencyLimit: 1, concurrencyLimit: 1,
}, },
concurrencyKey: data.userId, concurrencyKey: data.userId,
},
}); });
//return a success response with the handle //return a success response with the handle
return Response.json(handle); return Response.json(handle);
} else { } else {
//trigger the task, with a different queue //trigger the task, with a different queue
const handle = await generatePullRequest.trigger({ const handle = await generatePullRequest.trigger(data, {
payload: data,
options: {
queue: { queue: {
//every paid user gets a queue with a concurrency limit of 10 //every paid user gets a queue with a concurrency limit of 10
name: "paid-users", name: "paid-users",
concurrencyLimit: 10, concurrencyLimit: 10,
}, },
concurrencyKey: data.userId, concurrencyKey: data.userId,
},
}); });
//return a success response with the handle //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() { async function triggerHelloWorld() {
//This triggers the task and return a handle //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. //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); console.log("Task is running with handle", handle.id);
docs/v3/triggering.mdx
+33 -26
View File
@@ -48,7 +48,7 @@ export async function POST(request: Request) {
const data = await request.json(); const data = await request.json();
//trigger your task //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 a success response with the handle
return Response.json(handle); return Response.json(handle);
@@ -67,7 +67,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
const data = await request.json(); const data = await request.json();
//trigger your task //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 a success response with the handle
return json(handle); return json(handle);
@@ -91,9 +91,9 @@ export async function POST(request: Request) {
const data = await request.json(); const data = await request.json();
//batch trigger your task //batch trigger your task
const batchHandle = await emailSequence.batchTrigger({ const batchHandle = await emailSequence.batchTrigger(
items: data.users.map((u) => ({ payload: { to: u.email, name: u.name } })), data.users.map((u) => ({ payload: { to: u.email, name: u.name } }))
}); );
//return a success response with the handle //return a success response with the handle
return Response.json(batchHandle); return Response.json(batchHandle);
@@ -112,9 +112,9 @@ export async function action({ request, params }: ActionFunctionArgs) {
const data = await request.json(); const data = await request.json();
//batch trigger your task //batch trigger your task
const batchHandle = await emailSequence.batchTrigger({ const batchHandle = await emailSequence.batchTrigger(
items: data.users.map((u) => ({ payload: { to: u.email, name: u.name } })), data.users.map((u) => ({ payload: { to: u.email, name: u.name } }))
}); );
//return a success response with the handle //return a success response with the handle
return json(batchHandle); return json(batchHandle);
@@ -137,7 +137,7 @@ import { myOtherTask } from "~/trigger/my-other-task";
export const myTask = task({ export const myTask = task({
id: "my-task", id: "my-task",
run: async (payload: string) => { run: async (payload: string) => {
const handle = await myOtherTask.trigger({ payload: "some data" }); const handle = await myOtherTask.trigger("some data");
//...do other stuff //...do other stuff
}, },
@@ -154,7 +154,7 @@ import { myOtherTask } from "~/trigger/my-other-task";
export const myTask = task({ export const myTask = task({
id: "my-task", id: "my-task",
run: async (payload: string) => { run: async (payload: string) => {
const batchHandle = await myOtherTask.batchTrigger({ items: [{ payload: "some data" }] }); const batchHandle = await myOtherTask.batchTrigger([{ payload: "some data" }]);
//...do other stuff //...do other stuff
}, },
@@ -171,13 +171,15 @@ This is where it gets interesting. You can trigger a task and then wait for the
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> <CodeGroup>
```ts /trigger/batch.ts ```ts /trigger/batch.ts
export const batchTask = task({ export const batchTask = task({
id: "batch-task", id: "batch-task",
run: async (payload: string) => { run: async (payload: string) => {
const results = await childTask.batchTriggerAndWait({ const results = await childTask.batchTriggerAndWait([
items: [{ payload: "item1" }, { payload: "item2" }], { payload: "item1" },
}); { payload: "item2" },
]);
console.log("Results", results); console.log("Results", results);
//...do stuff with the results //...do stuff with the results
@@ -192,7 +194,7 @@ export const loopTask = task({
//this will be slower than the batch version //this will be slower than the batch version
//as we have to resume the parent after each iteration //as we have to resume the parent after each iteration
for (let i = 0; i < 2; i++) { 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); console.log("Result", result);
//...do stuff with the result //...do stuff with the result
@@ -200,6 +202,7 @@ export const loopTask = task({
}, },
}); });
``` ```
</CodeGroup> </CodeGroup>
</Accordion> </Accordion>
@@ -208,7 +211,7 @@ export const loopTask = task({
export const parentTask = task({ export const parentTask = task({
id: "parent-task", id: "parent-task",
run: async (payload: string) => { run: async (payload: string) => {
const result = await batchChildTask.triggerAndWait({ payload: "some-data" }); const result = await batchChildTask.triggerAndWait("some-data");
console.log("Result", result); console.log("Result", result);
//...do stuff with the result //...do stuff with the result
@@ -226,13 +229,15 @@ You can batch trigger a task and wait for all the results. This is useful for th
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> <CodeGroup>
```ts /trigger/batch.ts ```ts /trigger/batch.ts
export const batchTask = task({ export const batchTask = task({
id: "batch-task", id: "batch-task",
run: async (payload: string) => { run: async (payload: string) => {
const results = await childTask.batchTriggerAndWait({ const results = await childTask.batchTriggerAndWait([
items: [{ payload: "item1" }, { payload: "item2" }], { payload: "item1" },
}); { payload: "item2" },
]);
console.log("Results", results); console.log("Results", results);
//...do stuff with the results //...do stuff with the results
@@ -247,9 +252,10 @@ export const loopTask = task({
//this will be slower than a single batchTriggerAndWait() //this will be slower than a single batchTriggerAndWait()
//as we have to resume the parent after each iteration //as we have to resume the parent after each iteration
for (let i = 0; i < 2; i++) { for (let i = 0; i < 2; i++) {
const result = await childTask.batchTriggerAndWait({ const result = await childTask.batchTriggerAndWait([
items: [{ payload: `itemA${i}` }, { payload: `itemB${i}` }], { payload: `itemA${i}` },
}); { payload: `itemB${i}` },
]);
console.log("Result", result); console.log("Result", result);
//...do stuff with the result //...do stuff with the result
@@ -257,6 +263,7 @@ export const loopTask = task({
}, },
}); });
``` ```
</CodeGroup> </CodeGroup>
</Accordion> </Accordion>
@@ -265,9 +272,11 @@ export const loopTask = task({
export const batchParentTask = task({ export const batchParentTask = task({
id: "parent-task", id: "parent-task",
run: async (payload: string) => { run: async (payload: string) => {
const results = await childTask.batchTriggerAndWait({ const results = await childTask.batchTriggerAndWait([
items: [{ payload: "item4" }, { payload: "item5" }, { payload: "item6" }], { payload: "item4" },
}); { payload: "item5" },
{ payload: "item6" },
]);
console.log("Results", results); console.log("Results", results);
//...do stuff with the result //...do stuff with the result
@@ -326,9 +335,7 @@ import { createAvatar } from "@/trigger/create-avatar";
export async function create() { export async function create() {
try { try {
const handle = await createAvatar.trigger({ const handle = await createAvatar.trigger({
payload: {
userImage: "http://...", userImage: "http://...",
},
}); });
return { handle }; return { handle };
docs/v3/upgrading-from-v2.mdx
-2
View File
@@ -164,9 +164,7 @@ We've unified triggering in v3. You use `trigger()` or `batchTrigger()` which yo
async function yourBackendFunction() { async function yourBackendFunction() {
//call `trigger()` on any task //call `trigger()` on any task
const handle = await openaiTask.trigger({ 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
+3 -14
View File
@@ -8,10 +8,7 @@ import { RuntimeManager } from "./manager";
import { unboundedTimeout } from "../utils/timers"; import { unboundedTimeout } from "../utils/timers";
export class DevRuntimeManager implements RuntimeManager { export class DevRuntimeManager implements RuntimeManager {
_taskWaits: Map< _taskWaits: Map<string, { resolve: (value: TaskRunExecutionResult) => void }> = new Map();
string,
{ resolve: (value: TaskRunExecutionResult) => void; reject?: (err?: any) => void }
> = new Map();
_batchWaits: Map< _batchWaits: Map<
string, string,
@@ -41,8 +38,8 @@ export class DevRuntimeManager implements RuntimeManager {
return pendingCompletion; return pendingCompletion;
} }
const promise = new Promise<TaskRunExecutionResult>((resolve, reject) => { const promise = new Promise<TaskRunExecutionResult>((resolve) => {
this._taskWaits.set(params.id, { resolve, reject }); this._taskWaits.set(params.id, { resolve });
}); });
return await promise; return await promise;
@@ -93,15 +90,7 @@ export class DevRuntimeManager implements RuntimeManager {
return; return;
} }
if (!wait.reject) {
wait.resolve(completion); wait.resolve(completion);
} else {
if (completion.ok) {
wait.resolve(completion);
} else {
wait.reject(completion);
}
}
this._taskWaits.delete(execution.run.id); this._taskWaits.delete(execution.run.id);
} }
packages/core/src/v3/runtime/prodRuntimeManager.ts
+4 -16
View File
@@ -1,4 +1,3 @@
import { setTimeout } from "node:timers/promises";
import { clock } from "../clock-api"; import { clock } from "../clock-api";
import { import {
BatchTaskRunExecutionResult, BatchTaskRunExecutionResult,
@@ -8,19 +7,16 @@ import {
TaskRunExecution, TaskRunExecution,
TaskRunExecutionResult, TaskRunExecutionResult,
} from "../schemas"; } from "../schemas";
import { unboundedTimeout } from "../utils/timers";
import { ZodIpcConnection } from "../zodIpc"; import { ZodIpcConnection } from "../zodIpc";
import { RuntimeManager } from "./manager"; import { RuntimeManager } from "./manager";
import { unboundedTimeout } from "../utils/timers";
export type ProdRuntimeManagerOptions = { export type ProdRuntimeManagerOptions = {
waitThresholdInMs?: number; waitThresholdInMs?: number;
}; };
export class ProdRuntimeManager implements RuntimeManager { export class ProdRuntimeManager implements RuntimeManager {
_taskWaits: Map< _taskWaits: Map<string, { resolve: (value: TaskRunExecutionResult) => void }> = new Map();
string,
{ resolve: (value: TaskRunExecutionResult) => void; reject?: (err?: any) => void }
> = new Map();
_batchWaits: Map< _batchWaits: Map<
string, string,
@@ -91,8 +87,8 @@ export class ProdRuntimeManager implements RuntimeManager {
} }
async waitForTask(params: { id: string; ctx: TaskRunContext }): Promise<TaskRunExecutionResult> { async waitForTask(params: { id: string; ctx: TaskRunContext }): Promise<TaskRunExecutionResult> {
const promise = new Promise<TaskRunExecutionResult>((resolve, reject) => { const promise = new Promise<TaskRunExecutionResult>((resolve) => {
this._taskWaits.set(params.id, { resolve, reject }); this._taskWaits.set(params.id, { resolve });
}); });
await this.ipc.send("WAIT_FOR_TASK", { await this.ipc.send("WAIT_FOR_TASK", {
@@ -139,15 +135,7 @@ export class ProdRuntimeManager implements RuntimeManager {
return; return;
} }
if (!wait.reject) {
wait.resolve(completion); wait.resolve(completion);
} else {
if (completion.ok) {
wait.resolve(completion);
} else {
wait.reject(completion);
}
}
this._taskWaits.delete(execution.run.id); 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; 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. */ /** An id for your task. This must be unique inside your project and not change between versions. */
id: string; id: string;
/** The retry settings when an uncaught error is thrown. /** The retry settings when an uncaught error is thrown.
@@ -168,7 +172,7 @@ export type TaskRunResult<TOutput = any> =
| { | {
ok: false; ok: false;
id: string; id: string;
error: any; error: unknown;
}; };
export type BatchResult<TOutput = any> = { export type BatchResult<TOutput = any> = {
@@ -176,18 +180,72 @@ export type BatchResult<TOutput = any> = {
runs: TaskRunResult<TOutput>[]; 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; id: string;
trigger: (params: { payload: TInput; options?: TaskRunOptions }) => Promise<InvokeHandle>; /**
batchTrigger: (params: { * 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.
items: { payload: TInput; options?: TaskRunOptions }[]; * @param payload
// batchOptions?: BatchRunOptions; * @param options
}) => Promise<InvokeBatchHandle>; * @returns InvokeHandle
triggerAndWait: (params: { payload: TInput; options?: TaskRunOptions }) => Promise<TOutput>; * - `id` - The id of the triggered task run.
batchTriggerAndWait: (params: { */
items: { payload: TInput; options?: TaskRunOptions }[]; trigger: (payload: TInput, options?: TaskRunOptions) => Promise<InvokeHandle>;
// batchOptions?: BatchRunOptions;
}) => Promise<BatchResult<TOutput>>; /**
* 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 = { type TaskRunOptions = {
@@ -201,10 +259,6 @@ type TaskRunOptions = {
type TaskRunConcurrencyOptions = Queue; type TaskRunConcurrencyOptions = Queue;
type BatchRunOptions = TaskRunOptions & {
maxConcurrency?: number;
};
export type Prettify<T> = { export type Prettify<T> = {
[K in keyof T]: T[K]; [K in keyof T]: T[K];
} & {}; } & {};
@@ -213,12 +267,12 @@ export type DynamicBaseOptions = {
id: string; 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> params: TaskOptions<TInput, TOutput, TInitOutput>
): Task<TInput, TOutput> { ): Task<TInput, TOutput> {
const task: Task<TInput, TOutput> = { const task: Task<TInput, TOutput> = {
id: params.id, id: params.id,
trigger: async ({ payload, options }) => { trigger: async (payload, options) => {
const apiClient = apiClientManager.client; const apiClient = apiClientManager.client;
if (!apiClient) { if (!apiClient) {
@@ -258,7 +312,6 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
[SemanticInternalAttributes.STYLE_ICON]: "trigger", [SemanticInternalAttributes.STYLE_ICON]: "trigger",
["messaging.client_id"]: taskContextManager.worker?.id, ["messaging.client_id"]: taskContextManager.worker?.id,
[SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id, [SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.id,
["messaging.message.body.size"]: JSON.stringify(payload).length,
[SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev", [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
...(taskMetadata ...(taskMetadata
? accessoryAttributes({ ? accessoryAttributes({
@@ -277,7 +330,7 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
return handle; return handle;
}, },
batchTrigger: async ({ items }) => { batchTrigger: async (items) => {
const apiClient = apiClientManager.client; const apiClient = apiClientManager.client;
if (!apiClient) { if (!apiClient) {
@@ -323,9 +376,6 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
["messaging.batch.message_count"]: items.length, ["messaging.batch.message_count"]: items.length,
["messaging.client_id"]: taskContextManager.worker?.id, ["messaging.client_id"]: taskContextManager.worker?.id,
[SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.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", [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
[SemanticInternalAttributes.STYLE_ICON]: "trigger", [SemanticInternalAttributes.STYLE_ICON]: "trigger",
...(taskMetadata ...(taskMetadata
@@ -345,7 +395,7 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
return response; return response;
}, },
triggerAndWait: async ({ payload, options }) => { triggerAndWait: async (payload, options) => {
const ctx = taskContextManager.ctx; const ctx = taskContextManager.ctx;
if (!ctx) { if (!ctx) {
@@ -393,13 +443,7 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
} }
); );
const runResult = await handleTaskRunExecutionResult<TOutput>(result); return await handleTaskRunExecutionResult<TOutput>(result);
if (!runResult.ok) {
throw runResult.error;
}
return runResult.output;
} }
} }
@@ -408,13 +452,7 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
ctx, ctx,
}); });
const runResult = await handleTaskRunExecutionResult<TOutput>(result); return await handleTaskRunExecutionResult<TOutput>(result);
if (!runResult.ok) {
throw runResult.error;
}
return runResult.output;
}, },
{ {
kind: SpanKind.PRODUCER, 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; const ctx = taskContextManager.ctx;
if (!ctx) { if (!ctx) {
@@ -555,9 +593,6 @@ export function createTask<TInput, TOutput, TInitOutput extends InitOutput>(
["messaging.batch.message_count"]: items.length, ["messaging.batch.message_count"]: items.length,
["messaging.client_id"]: taskContextManager.worker?.id, ["messaging.client_id"]: taskContextManager.worker?.id,
[SEMATTRS_MESSAGING_DESTINATION]: params.queue?.name ?? params.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", [SEMATTRS_MESSAGING_SYSTEM]: "trigger.dev",
[SemanticInternalAttributes.STYLE_ICON]: "trigger", [SemanticInternalAttributes.STYLE_ICON]: "trigger",
...(taskMetadata ...(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 * @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> options: TaskOptions<TInput, TOutput, TInitOutput>
): Task<TInput, TOutput> { ): Task<TInput, TOutput> {
return createTask<TInput, TOutput, TInitOutput>(options); 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({ export const batchParentTask = task({
id: "batch-parent-task", id: "batch-parent-task",
run: async () => { run: async () => {
const response = await batchChildTask.batchTrigger({ const response = await batchChildTask.batchTrigger([
items: [{ payload: "item1" }, { payload: "item2" }, { payload: "item3" }], { payload: "item1" },
}); { payload: "item2" },
{ payload: "item3" },
]);
logger.info("Batch task response", { response }); logger.info("Batch task response", { response });
await wait.for({ seconds: 5 }); await wait.for({ seconds: 5 });
await wait.until({ date: new Date(Date.now() + 1000 * 5) }); // 5 seconds await wait.until({ date: new Date(Date.now() + 1000 * 5) }); // 5 seconds
const waitResponse = await batchChildTask.batchTriggerAndWait({ const waitResponse = await batchChildTask.batchTriggerAndWait([
items: [{ payload: "item4" }, { payload: "item5" }, { payload: "item6" }], { payload: "item4" },
}); { payload: "item5" },
{ payload: "item6" },
]);
logger.info("Batch task wait response", { waitResponse }); 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 new Promise((resolve) => setTimeout(resolve, 3000));
await testConcurrencyChild.batchTrigger({ await testConcurrencyChild.batchTrigger(
items: Array.from({ length: count }).map((_, index) => ({ Array.from({ length: count }).map((_, index) => ({
payload: { payload: {
delay, delay,
}, },
})), }))
}); );
logger.info(`All ${count} tasks triggered`); 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({ export const idempotencyKeyParent = task({
id: "idempotency-key-parent", id: "idempotency-key-parent",
run: async (payload: { key: string }) => { run: async (payload: { key: string }) => {
console.log("Hello from idempotency-key-parent"); console.log("Hello from idempotency-key-parent");
const childTaskResponse = await idempotencyKeyChild.triggerAndWait({ const childTaskResponse = await idempotencyKeyChild.triggerAndWait(
payload: { {
key: payload.key, key: payload.key,
forceError: true, forceError: true,
}, },
options: { {
idempotencyKey: payload.key, idempotencyKey: payload.key,
}, }
}); );
if (childTaskResponse.ok) {
logger.log("Child task response", { output: childTaskResponse.output });
} else {
logger.error("Child task error", { error: childTaskResponse.error });
}
return { return {
key: payload.key, key: payload.key,
@@ -42,8 +48,8 @@ export const idempotencyKeyBatchParent = task({
run: async (payload: { keyPrefix: string; itemCount: number }) => { run: async (payload: { keyPrefix: string; itemCount: number }) => {
console.log("Hello from idempotency-key-batch-parent"); console.log("Hello from idempotency-key-batch-parent");
const childTaskResponse = await idempotencyKeyBatchChild.batchTriggerAndWait({ const childTaskResponse = await idempotencyKeyBatchChild.batchTriggerAndWait(
items: Array.from({ length: payload.itemCount }).map((_, index) => ({ Array.from({ length: payload.itemCount }).map((_, index) => ({
payload: { payload: {
key: `${payload.keyPrefix}-${index}`, key: `${payload.keyPrefix}-${index}`,
forceError: index % 2 === 0, forceError: index % 2 === 0,
@@ -52,8 +58,8 @@ export const idempotencyKeyBatchParent = task({
options: { options: {
idempotencyKey: `${payload.keyPrefix}-${index}`, idempotencyKey: `${payload.keyPrefix}-${index}`,
}, },
})), }))
}); );
return { return {
keyPrefix: payload.keyPrefix, 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 }) => { run: async (payload: { message: string }) => {
logger.info("Long running parent", { payload }); logger.info("Long running parent", { payload });
await longRunning.triggerAndWait({ payload: { message: "child" } }); await longRunning.triggerAndWait({ message: "child" });
return { return {
finished: new Date().toISOString(), finished: new Date().toISOString(),
references/v3-catalog/src/trigger/simple.ts
-4
View File
@@ -83,19 +83,15 @@ export const parentTask = task({
await wait.for({ seconds: 5 }); await wait.for({ seconds: 5 });
const childTaskResponse = await childTask.triggerAndWait({ const childTaskResponse = await childTask.triggerAndWait({
payload: {
message: payload.message, message: payload.message,
forceError: false, forceError: false,
},
}); });
logger.info("Child task response", { childTaskResponse }); logger.info("Child task response", { childTaskResponse });
await childTask.trigger({ await childTask.trigger({
payload: {
message: `${payload.message} - 2.a`, message: `${payload.message} - 2.a`,
forceError: true, forceError: true,
},
}); });
return { return {
references/v3-catalog/src/trigger/subtasks.ts
+33 -44
View File
@@ -5,36 +5,28 @@ export const simpleParentTask = task({
id: "simple-parent-task", id: "simple-parent-task",
run: async (payload: { message: string }) => { run: async (payload: { message: string }) => {
await simpleChildTask.trigger({ await simpleChildTask.trigger({
payload: {
message: `${payload.message} - 2.a`, message: `${payload.message} - 2.a`,
},
}); });
await simpleChildTask.triggerAndWait({ await simpleChildTask.triggerAndWait({
payload: {
message: `${payload.message} - 2.b`, message: `${payload.message} - 2.b`,
},
}); });
await simpleChildTask.batchTrigger({ await simpleChildTask.batchTrigger([
items: [
{ {
payload: { payload: {
message: `${payload.message} - 2.c`, message: `${payload.message} - 2.c`,
}, },
}, },
], ]);
});
await simpleChildTask.batchTriggerAndWait({ await simpleChildTask.batchTriggerAndWait([
items: [
{ {
payload: { payload: {
message: `${payload.message} - 2.d`, message: `${payload.message} - 2.d`,
}, },
}, },
], ]);
});
return { return {
hello: "world", hello: "world",
@@ -53,13 +45,10 @@ export const subtasksWithRetries = task({
id: "subtasks-with-retries", id: "subtasks-with-retries",
run: async (payload: { message: string }) => { run: async (payload: { message: string }) => {
await taskWithRetries.triggerAndWait({ await taskWithRetries.triggerAndWait({
payload: {
message: `${payload.message} - 2.b`, message: `${payload.message} - 2.b`,
},
}); });
await taskWithRetries.batchTrigger({ await taskWithRetries.batchTrigger([
items: [
{ {
payload: { payload: {
message: `${payload.message} - 2.c`, message: `${payload.message} - 2.c`,
@@ -70,11 +59,9 @@ export const subtasksWithRetries = task({
message: `${payload.message} - 2.cc`, message: `${payload.message} - 2.cc`,
}, },
}, },
], ]);
});
await taskWithRetries.batchTriggerAndWait({ await taskWithRetries.batchTriggerAndWait([
items: [
{ {
payload: { payload: {
message: `${payload.message} - 2.d`, message: `${payload.message} - 2.d`,
@@ -85,17 +72,13 @@ export const subtasksWithRetries = task({
message: `${payload.message} - 2.dd`, message: `${payload.message} - 2.dd`,
}, },
}, },
], ]);
});
await taskWithRetries.triggerAndWait({ await taskWithRetries.triggerAndWait({
payload: {
message: `${payload.message} - 2.e`, message: `${payload.message} - 2.e`,
},
}); });
await taskWithRetries.batchTriggerAndWait({ await taskWithRetries.batchTriggerAndWait([
items: [
{ {
payload: { payload: {
message: `${payload.message} - 2.f`, message: `${payload.message} - 2.f`,
@@ -106,8 +89,7 @@ export const subtasksWithRetries = task({
message: `${payload.message} - 2.ff`, message: `${payload.message} - 2.ff`,
}, },
}, },
], ]);
});
return { return {
hello: "world", hello: "world",
@@ -118,21 +100,17 @@ export const subtasksWithRetries = task({
export const multipleTriggerWaits = task({ export const multipleTriggerWaits = task({
id: "multiple-trigger-waits", id: "multiple-trigger-waits",
run: async ({ message = "test" }: { message?: string }) => { run: async ({ message = "test" }: { message?: string }) => {
await simpleChildTask.triggerAndWait({ payload: { message: `${message} - 1.a` } }); await simpleChildTask.triggerAndWait({ message: `${message} - 1.a` });
await simpleChildTask.triggerAndWait({ payload: { message: `${message} - 2.a` } }); await simpleChildTask.triggerAndWait({ message: `${message} - 2.a` });
await simpleChildTask.batchTriggerAndWait({ await simpleChildTask.batchTriggerAndWait([
items: [
{ payload: { message: `${message} - 3.a` } }, { payload: { message: `${message} - 3.a` } },
{ payload: { message: `${message} - 3.b` } }, { payload: { message: `${message} - 3.b` } },
], ]);
}); await simpleChildTask.batchTriggerAndWait([
await simpleChildTask.batchTriggerAndWait({
items: [
{ payload: { message: `${message} - 4.a` } }, { payload: { message: `${message} - 4.a` } },
{ payload: { message: `${message} - 4.b` } }, { payload: { message: `${message} - 4.b` } },
], ]);
});
return { return {
hello: "world", hello: "world",
@@ -144,19 +122,21 @@ export const triggerAndWaitLoops = task({
id: "trigger-wait-loops", id: "trigger-wait-loops",
run: async ({ message = "test" }: { message?: string }) => { run: async ({ message = "test" }: { message?: string }) => {
for (let i = 0; i < 2; i++) { 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++) { for (let i = 0; i < 2; i++) {
await simpleChildTask.batchTriggerAndWait({ await simpleChildTask.batchTriggerAndWait([
items: [
{ payload: { message: `${message} - ${i}.a` } }, { payload: { message: `${message} - ${i}.a` } },
{ payload: { message: `${message} - ${i}.b` } }, { payload: { message: `${message} - ${i}.b` } },
], ]);
// batchOptions: { maxConcurrency: 1 },
});
} }
await taskWithNoPayload.trigger();
await taskWithNoPayload.triggerAndWait();
await taskWithNoPayload.batchTrigger([{}]);
await taskWithNoPayload.batchTriggerAndWait([{}]);
// Don't do this! // Don't do this!
// await Promise.all( // await Promise.all(
// [{ message: `${message} - 1` }, { message: `${message} - 2` }].map((payload) => // [{ 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
+14 -20
View File
@@ -4,19 +4,19 @@ export const superParentTask = task({
id: "super-parent-task", id: "super-parent-task",
run: async () => { run: async () => {
const result = await superChildTask.triggerAndWait({ const result = await superChildTask.triggerAndWait({
payload: {
foo: "bar", foo: "bar",
whenToDo: new Date(), whenToDo: new Date(),
},
}); });
logger.log(`typeof result.date = ${typeof result.date}`); if (result.ok) {
logger.log(`typeof result.regex = ${typeof result.regex}`); logger.log(`typeof result.date = ${typeof result.output.date}`);
logger.log(`typeof result.bigint = ${typeof result.bigint}`); logger.log(`typeof result.output.regex = ${typeof result.output.regex}`);
logger.log(`typeof result.set = ${typeof result.set}`); logger.log(`typeof result.output.bigint = ${typeof result.output.bigint}`);
logger.log(`typeof result.map = ${typeof result.map}`); logger.log(`typeof result.output.set = ${typeof result.output.set}`);
logger.log(`typeof result.error = ${typeof result.error}`); logger.log(`typeof result.output.map = ${typeof result.output.map}`);
logger.log(`typeof result.url = ${typeof result.url}`); 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"; return "## super-parent-task completed";
}, },
@@ -49,14 +49,11 @@ export const superHugePayloadTask = task({
run: async () => { run: async () => {
const largePayload = createLargeObject(1000, 128); const largePayload = createLargeObject(1000, 128);
const result = await superHugeOutputTask.triggerAndWait({ const result = await superHugeOutputTask.triggerAndWait(largePayload);
payload: largePayload,
});
logger.log("Result from superHugeOutputTask: ", { result }); logger.log("Result from superHugeOutputTask: ", { result });
const batchResult = await superHugeOutputTask.batchTriggerAndWait({ const batchResult = await superHugeOutputTask.batchTriggerAndWait([
items: [
{ payload: largePayload }, { payload: largePayload },
{ {
payload: { payload: {
@@ -105,8 +102,7 @@ export const superHugePayloadTask = task({
small: "object", small: "object",
}, },
}, },
], ]);
});
logger.log("Result from superHugeOutputTask batchTriggerAndWait: ", { batchResult }); logger.log("Result from superHugeOutputTask batchTriggerAndWait: ", { batchResult });
@@ -118,7 +114,7 @@ export const superHugePayloadTask = task({
export const superHugeOutputTask = task({ export const superHugeOutputTask = task({
id: "super-huge-output-task", id: "super-huge-output-task",
run: async (payload) => { run: async (payload: any) => {
return payload; return payload;
}, },
}); });
@@ -127,9 +123,7 @@ export const superStringTask = task({
id: "super-string-parent-task", id: "super-string-parent-task",
run: async () => { run: async () => {
const result = await superStringChildTask.triggerAndWait({ const result = await superStringChildTask.triggerAndWait({
payload: {
foo: "bar", foo: "bar",
},
}); });
return result; return result;
@@ -138,7 +132,7 @@ export const superStringTask = task({
export const superStringChildTask = task({ export const superStringChildTask = task({
id: "super-string-child-task", id: "super-string-child-task",
run: async () => { run: async (payload: any) => {
return "## super-string-child-task completed"; return "## super-string-child-task completed";
}, },
}); });