vtex-io-events-and-workers

事件、Worker 与异步处理

适用场景

当 VTEX IO 应用需要通过事件、Worker 或其他后台执行模式异步处理工作时，使用此技能。

• 消费其他 VTEX 服务广播的事件
• 执行不应阻塞 HTTP 响应的后台工作
• 设计可安全重试的处理程序
• 处理批次或延迟任务
• 构建与外部服务的异步集成

以下情况不要使用此技能：

• 定义 HTTP 路由契约
• 设计 GraphQL 架构或解析器
• 决定应用级策略
• 构建底层客户端

决策规则

• 当工作开销大、容易触发重试，或无需在请求-响应周期内完成时，使用事件或 Worker。
• VTEX 使用内部事件广播器向你的服务传递平台事件和应用事件。同一广播器也可以将你的应用发布的事件路由到其他处理程序。假设两个方向都采用至少一次投递语义：事件可能会重试或重放，因此处理程序必须具备幂等性，并能安全处理重复事件。
• 保持事件处理程序具备幂等性。同一事件可能会投递多次，因此处理程序必须能够安全地应对重放。
• 在适当的存储中持久化幂等性和处理状态，例如使用 VBase 保存键值标记，或使用 Master Data 保存结构化记录，以便处理程序在重试时检测重复事件、已完成的工作和失败情况。
• 在 service.json 中明确声明事件和 Worker，以便将其接入 IO 运行时；保持输入契约稳定且明确，不要依赖 HTTP 路由假设。
• 需要通知其他应用或分发工作时，通过适当的 VTEX IO 客户端或事件机制发布事件，而不是创建临时 HTTP 回调来模拟异步投递。
• 要通过 VTEX IO 事件总线发布事件，应用通常需要在 manifest.json 中配置 colossus-fire-event 策略。只有当应用也确实会使用这些受保护资源时，才添加其他策略。
• 当处理程序超出小型、明确的工作单元时，将事件接收与业务编排分开。
• 将重试视为预期行为，而非异常行为。设计处理程序，确保重复执行也安全。
• 明确说明后台处理程序产生的副作用，例如写入、外部调用或状态转换。
• 对于面向批次的处理程序，将工作拆分为小型、明确的单元，并记录每个项目的状态，以免单个失败项掩盖批次其余部分的进度。

硬性约束

约束：事件处理程序必须具备幂等性

每个事件或后台处理程序都必须能够容忍重复执行，且不会产生不一致的副作用。

原因

异步系统会重试。缺少幂等性时，重复投递可能导致重复创建记录、重复调用合作伙伴，或发生无效的状态转换。

检测

如果处理程序在执行写入或产生外部副作用时，没有检查工作是否已完成，请停止并先添加幂等性保护。

正确

export async function handleOrderCreated(ctx: Context) {
  const { orderId } = ctx.body
  const alreadyProcessed = await ctx.clients.statusStore.hasProcessed(orderId)

  if (alreadyProcessed) {
    return
  }

  await ctx.clients.partnerApi.sendOrder(orderId)
  await ctx.clients.statusStore.markProcessed(orderId)
}

错误

export async function handleOrderCreated(ctx: Context) {
  await ctx.clients.partnerApi.sendOrder(ctx.body.orderId)
}

约束：后台工作不得依赖仅适用于请求的假设

Worker 和事件处理程序不得依赖仅适用于 HTTP 的假设，例如路由参数、即时用户交互或绑定到请求的可变状态。

原因

异步处理程序在路由生命周期之外运行。沿用 HTTP 假设会导致上下文缺失、行为脆弱，以及同步与异步路径之间意外耦合。

检测

如果事件处理程序依赖请求标头、路由参数或特定路由的状态结构，请停止并重新设计输入契约，使处理程序接收明确的异步数据。

正确

export async function handleImport(ctx: Context) {
  const { importId, account } = ctx.body
  await ctx.clients.importApi.process(importId, account)
}

错误

export async function handleImport(ctx: Context) {
  await ctx.clients.importApi.process(ctx.vtex.route.params.id, ctx.request.header.account)
}

约束：耗时的异步流程必须清楚呈现部分失败

异步处理程序 MUST 通过状态、日志或持久化标记明确呈现部分失败，而不是悄悄将其吞掉。

为什么这很重要

后台失败比路由失败更难发现。如果没有明确的失败信号，运维团队就无法判断任务是被跳过、重试，还是只完成了一部分。

检测

如果处理程序捕获错误，却没有记录失败状态、记录足够的上下文，或在适当情况下重新抛出错误，请 STOP 并明确处理失败的方式。

正确

export async function handleSync(ctx: Context) {
  try {
    await ctx.clients.partnerApi.syncCatalog(ctx.body.catalogId)
    await ctx.clients.statusStore.markSuccess(ctx.body.catalogId)
  } catch (error) {
    await ctx.clients.statusStore.markFailure(ctx.body.catalogId)
    throw error
  }
}

错误

export async function handleSync(ctx: Context) {
  try {
    await ctx.clients.partnerApi.syncCatalog(ctx.body.catalogId)
  } catch (_) {
    return
  }
}

首选模式

推荐的文件布局：

node/
├── events/
│   ├── index.ts
│   ├── catalog.ts
│   └── orders.ts
└── workers/
    └── sync.ts

最简异步处理程序模式：

export async function handleCatalogChanged(ctx: Context) {
  const { skuId } = ctx.body
  const alreadyDone = await ctx.clients.syncState.isProcessed(skuId)

  if (alreadyDone) {
    return
  }

  await ctx.clients.catalogSync.syncSku(skuId)
  await ctx.clients.syncState.markProcessed(skuId)
}

事件发布示例模式：

export async function broadcast(ctx: Context, next: () => Promise<void>) {
  const {
    clients: { events },
    body: { payload, senderAppId, clientAppId },
  } = ctx

  for (const row of payload as unknown[]) {
    await events.sendEvent(clientAppId, 'my-app.event-name', {
      data: row,
      senderAppId,
    })
  }

  await next()
}

事件发布的最简清单策略：

{
  "policies": [
    {
      "name": "colossus-fire-event"
    }
  ]
}

使用路由确认或触发任务，并使用事件或工作进程执行耗时、可重复且能感知失败的处理。

有目的地使用存储来保存异步状态：

• VBase：使用外部标识符作为键，存储简单的幂等性标记
• Master Data：存储包含状态和时间戳的结构化处理记录

将异步负载视为独立的契约，不要沿用仅适用于路由的 HTTP 请求假设。

对于扇出或跨应用通知，发布包含 ID 和最少元数据、定义明确的小型事件，然后让下游处理程序在需要时从事实来源获取完整详情，而不是嵌入大型负载或依赖自定义回调 URL。

在开发过程中，使用 Broadcaster 应用在 Admin 中的 Notify Target Workspace 设置，将事件路由到特定工作区，而不是临时创建公共路由或仅用于测试的投递流程。无论哪个工作区接收事件，处理程序都应能正确运行。

常见故障模式

• 将事件投递视为恰好一次，而不是至少一次。
• 在工作进程或事件处理程序中沿用 HTTP 路由假设。
• 吞掉后台错误，却不明确记录失败状态。
• 让单个事件处理程序协调过多无关的副作用。
• 在路由中同步执行耗时任务，而不是将其移至异步处理。
• 记录完整事件负载（其中可能含有机密或令牌），而不是使用 ID 和元数据进行关联。

审查清单

• 异步处理是否适用于这项工作？
• 处理程序在重复投递时是否具备幂等性？
• 幂等性或处理状态是否存储在 VBase 或 Master Data 等合适的后端中？
• 是否在 service.json 中明确声明了事件和工作进程？
• 后台输入是否明确，且不依赖 HTTP 路由假设？
• 是否足够清楚地呈现失败，以便进行重试和故障排查？
• 对于批处理，是否能按项目或较小的工作单元查看状态？
• 是否应将大型处理程序拆分为更小的异步单元或编排步骤？

相关技能

• vtex-io-data-access-patterns - 用于选择幂等性键、同步状态或处理记录的存储位置
• vtex-io-observability-and-ops - 用于了解如何记录、度量和监控异步失败

参考资料

• Service - 事件声明和服务执行模型
• Node Builder - 服务的后端文件结构
• Broadcaster - 内部事件投递上下文，以及开发时使用的 Notify Target Workspace 设置