trigger-tasks

编写 Trigger.dev 任务

任务是能够长时间运行且具有很强故障恢复能力的函数。在 /trigger 目录下的文件中定义任务。始终从 @trigger.dev/sdk 导入。切勿从 @trigger.dev/sdk/v3（已弃用的别名）或 @trigger.dev/core 导入。

设置

// /trigger/hello-world.ts
import { task } from "@trigger.dev/sdk";

export const helloWorld = task({
  id: "hello-world", // unique within the project
  run: async (payload: { message: string }, { ctx }) => {
    console.log(payload.message, "attempt", ctx.attempt.number);
    return { ok: true }; // must be JSON serializable
  },
});

run 函数接收负载以及第二个参数，其中包含 ctx（运行上下文）、一个中止 signal 和一个已弃用的 init 输出。返回值是任务输出，且必须可序列化为 JSON。

核心模式

1. 使用 schemaTask 验证负载

schema 接受 Zod / Yup / Superstruct / ArkType / valibot / typebox 解析器，或自定义的 (data: unknown) => T 函数。验证失败时会抛出 TaskPayloadParsedError，并跳过重试。

import { schemaTask } from "@trigger.dev/sdk";
import { z } from "zod";

export const createUser = schemaTask({
  id: "create-user",
  schema: z.object({ name: z.string(), age: z.number() }),
  run: async (payload) => ({ greeting: `Hi ${payload.name}` }),
});

2. 配置重试并尽早中止

默认 maxAttempts 为 3。抛出 AbortTaskRunError 可立即停止重试。任务级别的 retry 会覆盖配置文件中的默认设置。

import { task, AbortTaskRunError } from "@trigger.dev/sdk";

export const charge = task({
  id: "charge",
  retry: { maxAttempts: 5, factor: 1.8, minTimeoutInMs: 500, maxTimeoutInMs: 30_000, randomize: true },
  run: async (payload: { amount: number }) => {
    if (payload.amount <= 0) throw new AbortTaskRunError("Invalid amount"); // no retry
    // work that may throw and retry
  },
});

如需更精细的控制，catchError: async ({ payload, error, ctx, retryAt }) => {...} 可以返回 { skipRetrying: true }、{ retryAt: Date } 或 undefined（使用常规逻辑）。任务内重试还可以使用 retry.onThrow、retry.fetch。

3. 触发另一个任务并处理 Result

在任务内部使用 yourTask.triggerAndWait(payload)。结果是一个 Result 对象，必须检查其 ok，或调用 .unwrap() 以在失败时抛出异常。

export const parentTask = task({
  id: "parent-task",
  run: async () => {
    const result = await childTask.triggerAndWait({ data: "x" });
    if (result.ok) return result.output; // typed child output
    console.error("child failed", result.error);
    // or: const output = await childTask.triggerAndWait({ data: "x" }).unwrap();
  },
});

SubtaskUnwrapError 包含 runId、taskId 和 cause。如需扇出，可使用 childTask.batchTriggerAndWait([{ payload: a }, { payload: b }])；结果包含一个 .runs 数组，每个条目为 { ok, id, output?, error?, taskIdentifier }。

4. 在后端代码中通过仅类型导入来触发任务

在任务外部，只导入任务类型，并通过 ID 触发任务。不要将任务实例导入后端打包文件。

import { tasks } from "@trigger.dev/sdk";
import type { emailSequence } from "~/trigger/emails";

const handle = await tasks.trigger<typeof emailSequence>(
  "email-sequence",
  { to: "[email protected]", name: "Ada" },
  { delay: "1h" }
);

批量触发可使用 tasks.batchTrigger 和 batch.trigger([{ id, payload }])。触发选项包括 delay、ttl、idempotencyKey、idempotencyKeyTTL、debounce、queue、concurrencyKey、maxAttempts、tags、metadata、priority、region 和 machine。使用 runs.retrieve、runs.cancel 和 runs.reschedule 检查运行。

5. 幂等键

idempotencyKeys.create(key, { scope }) 会返回一个经过哈希处理的 64 字符键。原始字符串键默认使用 "run" 作用域（v4.3.1+）；如需仅执行一次的行为，请使用 scope: "global"。

import { idempotencyKeys, task } from "@trigger.dev/sdk";

export const processOrder = task({
  id: "process-order",
  run: async (payload: { orderId: string; email: string }) => {
    const key = await idempotencyKeys.create(`confirm-${payload.orderId}`);
    await sendEmail.trigger({ to: payload.email }, { idempotencyKey: key });
  },
});

6. 等待和运行元数据

wait.for({ seconds }) 和 wait.until({ date }) 会持久化暂停运行。metadata.* 只能在 run() 内读取和写入；更新是同步的，并且支持链式调用（set、del、replace、append、remove、increment、decrement）。

import { task, metadata, wait } from "@trigger.dev/sdk";

export const importer = task({
  id: "importer",
  run: async (payload: { rows: unknown[] }) => {
    metadata.set("status", "processing").set("total", payload.rows.length);
    await wait.for({ seconds: 5 });
    metadata.set("status", "complete");
  },
});

对于人机协作流程，wait.createToken({ timeout, tags }) 会返回 { id, url, publicAccessToken, ... }；使用 wait.forToken<T>(token: string | { id: string }) 恢复运行，该方法返回 { ok, output?, error? }（或调用 .unwrap()），并可在其他位置通过 wait.completeToken(tokenId, output) 完成等待。元数据上限为 256KB，且不会传递给子任务；如需将值传递给父任务，请使用 metadata.parent.* / metadata.root.*。（自 4.1.0 起，metadata.stream 已弃用，建议改用 streams.pipe()。）

7. 定时（cron）任务

import { schedules } from "@trigger.dev/sdk";

export const dailyReport = schedules.task({
  id: "daily-report",
  cron: { pattern: "0 5 * * *", timezone: "Asia/Tokyo" },
  run: async (payload) => {
    console.log("scheduled at", payload.timestamp, "next", payload.upcoming);
  },
});

负载包含 timestamp、lastTimestamp、timezone、scheduleId、externalId 和 upcoming。使用 schedules.create({ task, cron, timezone?, externalId?, deduplicationKey }) 动态附加计划（去重键为必填项，且按项目区分），并可使用 retrieve / list / update / activate / deactivate / del / timezones。

8. 队列和并发

在任务上设置 queue: { concurrencyLimit }，或在多个任务之间共享队列：

import { queue, task } from "@trigger.dev/sdk";

export const emails = queue({ name: "emails", concurrencyLimit: 5 });

export const sendEmail = task({ id: "send-email", queue: emails, run: async () => {} });

触发时可使用 { queue: "queue-name" } 覆盖队列，并添加 concurrencyKey 以实现按租户划分的队列。使用 queues.list / retrieve / pause / resume / overrideConcurrencyLimit / resetConcurrencyLimit 管理队列。

9. trigger.config.ts 要点

import { defineConfig } from "@trigger.dev/sdk";

export default defineConfig({
  project: "<project ref>",
  dirs: ["./trigger"],
  machine: "small-1x",
  retries: {
    enabledInDev: false,
    default: { maxAttempts: 3, factor: 2, minTimeoutInMs: 1000, maxTimeoutInMs: 10000, randomize: true },
  },
});

build.external 用于指定哪些包不纳入打包。构建扩展（additionalFiles、prismaExtension、puppeteer、playwright、ffmpeg、pythonExtension、aptGet、syncEnvVars 等）来自 @trigger.dev/build。telemetry 用于配置插桩和导出器。每个扩展都有自己的设置文档，均位于 @trigger.dev/sdk/docs/config/extensions/（从 overview.mdx 开始）；请在接入前阅读所需文档，不要猜测 API。

日志记录

logger.debug / log / info / warn / error(message, dataRecord?) 会写入结构化日志；logger.trace(name, async (span) => {...}) 会添加一个 span。模块级指标使用 otel.metrics.getMeter(name)。

常见错误

1. 严重：将等待结果当作输出。 triggerAndWait 和 wait.forToken 返回的是 Result 对象，而不是原始输出。

   • 错误：const out = await childTask.triggerAndWait(p); use(out.foo);
   • 正确：const r = await childTask.triggerAndWait(p); if (r.ok) use(r.output.foo);（或使用 .unwrap()）。

2. 用 Promise.all 包装 triggerAndWait / batchTriggerAndWait / wait。

   • 错误：await Promise.all([childTask.triggerAndWait(a), childTask.triggerAndWait(b)]);
   • 正确：await childTask.batchTriggerAndWait([{ payload: a }, { payload: b }]);（或使用顺序执行的 for 循环）。

3. 在后端代码中导入任务实例。

   • 错误：在路由处理程序中使用 import { emailSequence } from "~/trigger/emails";。
   • 正确：使用 import type { emailSequence }，再调用 tasks.trigger<typeof emailSequence>("email-sequence", payload)。

4. 在 run() 外调用 metadata.set/get。

   • 错误：在模块作用域或无关的后端代码中设置元数据（不会执行任何操作；get 会返回 undefined）。
   • 正确：在 run() 或任务生命周期钩子中调用。

5. 假定子任务会继承父任务的队列或元数据。

   • 错误：期望子任务与父任务共用 concurrencyLimit，或能读取父任务的元数据。
   • 正确：子任务在自己的队列上运行；通过 { metadata: metadata.current() } 显式传递元数据，或使用 metadata.parent.* 向上推送。

6. 将原生/WASM 包纳入打包。

   • 错误：让 sharp、re2、sqlite3 或 WASM 包留在默认打包中。
   • 正确：将它们添加到 trigger.config.ts 中的 build.external。

7. 依赖原始字符串幂等键具备全局作用域。

   • 错误：使用 trigger(p, { idempotencyKey: "welcome-email" }) 并期望该操作永久只执行一次（仅在 v4.3.0 及更早版本中成立）。
   • 正确：await idempotencyKeys.create("welcome-email", { scope: "global" })。

参考资料

相关技能：

• trigger-realtime：订阅运行，并使用 React hooks 从前端触发任务。
• trigger-authoring-chat-agent 和 trigger-chat-agent-advanced：构建 AI 聊天 Agent。

参考文档与此技能一同包含在同一个包中，请在本地阅读（无需网络），文档版本与已安装版本一致。上方的 sources: frontmatter 列出了此技能所参考的所有文档，均位于 @trigger.dev/sdk/docs/ 下。请从以下文档开始：

• @trigger.dev/sdk/docs/tasks/overview.mdx
• @trigger.dev/sdk/docs/triggering.mdx
• @trigger.dev/sdk/docs/config/config-file.mdx

版本

此技能包含在 @trigger.dev/sdk 中，并直接从 node_modules 读取，因此始终与已安装的 SDK 版本一致（参见相邻的 package.json）。这些 API 的完整文档也随包一同提供，位于 @trigger.dev/sdk/docs/ 下。