workflow

@falcondev-oss/workflow

一个长期运行的异步进程——发送电子邮件，等待三天，再发送一封跟进邮件——会随运行它的进程一起终止。这个库让它能够在进程终止后继续运行：工作流主体是普通的异步函数，但它执行的每个 step 都会持久化到 Redis，因此发生崩溃、部署或重试时，会从头重放该函数，并直接跳过已完成的步骤。

共有三层，每一层都由上一层创建：

层	管理内容
WorkflowNamespace	Redis 连接、发布/订阅连接，以及跨工作流的并发上限
WorkflowDefinition	一个队列和一种任务类型：输入架构和选项。负责入队、安排任务、检查任务
Workflow	一个定义及其 run 处理程序，因此也能调用 `work()``
WorkflowJob	一次已入队的运行：id、groupId、wait()

所有内容都从软件包根目录导出：WorkflowNamespace、createRedis、expBackoff、TimeoutError、ResultExpiredError、JobAlreadyExistsError、NonRecoverableError。始终通过命名空间（createWorkflow / defineWorkflow）创建工作流，不要使用 new Workflow()。

设置

import { createRedis, WorkflowNamespace } from '@falcondev-oss/workflow'

const namespace = new WorkflowNamespace({
  id: 'app',
  redis: await createRedis({ url: process.env.REDIS_URL }),
  logger: console, // opt-in; no logging by default
})

命名空间负责所有配置：redis、logger（info/success/error/debug/warn 均为可选），以及工作流继承的选项集合。未提供 redis 时，库会回退到进程级的 localhost:6379 连接。createRedis 会应用库所需的连接选项（maxRetriesPerRequest: null、关闭离线队列、限制重试策略），因此应使用它创建连接，而不是使用 new IORedis()。

命名空间会注册自己的退出钩子，因此 SIGTERM 会排空工作进程并断开连接。在测试中，传入 autoClose: false 并自行调用 namespace.close()。

定义工作流

const onboarding = namespace.createWorkflow({
  id: 'onboarding',
  schema: z.object({ userId: z.string(), name: z.string() }),
  getGroupId: (input) => input.userId,
  async run({ input, step }) {
    await step.do('welcome', () => sendWelcome(input.userId))
    await step.wait('settle', 3 * 24 * 60 * 60 * 1000)
    const engaged = await step.do('check engagement', () => isEngaged(input.userId))
    return { engaged }
  },
})

• id 为此工作流的每个 Redis 键划定作用域。更改它会使正在运行的任务失去关联。
• schema 可以是任意 Standard Schema（zod、arktype 等），并会进行两次验证：入队时一次，工作进程中再一次。验证失败会抛出 Invalid workflow input。省略它时，input 的类型为 undefined。
• getGroupId 根据已验证的输入生成组 ID；运行调用中显式指定的 groupId 优先于它。
• run 接收 { input, step, span }，其解析后的值会成为任务输出。负载、步骤结果和输出都会通过 superjson 往返序列化，因此 Date、Map、Set 和 BigInt 都能完整保留。
• queueOptions / workerOptions / jobOptions 用于设置每个工作流的默认值；命名空间中相同的选项会在其下方进行浅合并。请参阅 reference/options.md。

定义和处理程序

createWorkflow 会将处理程序放在工作流的模块中，因此每个将其加入队列的模块也会加载处理程序导入的所有内容。如果加入队列的代码需要保持轻量（客户端模块、导入循环、缓慢的测试启动），请将其拆分：

// jobs/onboarding.ts: imported wherever jobs are enqueued
export const onboarding = namespace
  .defineWorkflow({ id: 'onboarding', schema: z.object({ userId: z.string() }) })
  .returns<{ engaged: boolean }>()

// handlers/onboarding.ts: imported only by the worker process
export default onboarding.handle(async ({ input, step }) => ({ engaged: await check(input.userId) }))

// worker.ts
import onboarding from './handlers/onboarding'
await onboarding.work()

• defineWorkflow 接受除 run 以外的所有 createWorkflow 选项。定义包含除 work() 以外的所有工作流方法。
• returns<T>() 声明输出类型，以便加入队列的代码中的 job.wait() 具有类型。若不使用它，定义的输出类型为 unknown。
• handle(run) 返回完整的 Workflow。其处理程序必须返回声明的类型；若不使用 returns<T>()，则根据处理程序推断输出类型。
• 对于已有处理程序的工作流，调用 handle() 和 returns() 会产生类型错误。该库不会对处理程序使用 import()，因此 worker 入口必须导入它所服务的每个处理程序模块。

await onboarding.work() // start a worker in this process

const job = await onboarding.run({ userId: 'u1', name: 'John' })
const { engaged } = await job.wait() // typed as the run() return

run(input, opts?) 会立即加入队列，runIn(input, delayMs, opts?) 会在延迟后加入队列，runAt(input, date, opts?) 会在指定时间戳加入队列。这三者都会在任务加入队列后立即解析为 WorkflowJob，不会等待任务执行。opts 接受 groupId、priority、maxAttempts 和 jobId。

job.wait(timeoutMs?) 会等待 pub/sub 通知，并返回反序列化后的输出。即使进程中没有 worker，纯生产者也可以使用它。它会因以下原因而拒绝：

错误	时机
工作流自身的 Error	run 抛出错误且已用尽重试次数时——原始消息会保留
TimeoutError	timeoutMs 先到期时（没有默认值；不传时会一直等待）
ResultExpiredError	任务已完成，但其结果记录因 resultTtl 到期而被清除时（默认值为 300 s）

如果传入的 jobId 已在使用中，则会因 JobAlreadyExistsError 而拒绝——这是用于“最多加入队列一次”的去重句柄。

步骤和重放

step.do(name, run) 是持久化的基本单位。它会在完成时缓存结果：一旦返回，结果就会写入任务的步骤哈希中，之后的任何重试都会返回缓存值，而不会重新运行函数体。因此，重试会立即重放已完成的步骤，只重新执行失败的步骤。

const user = await step.do('load user', () => db.user.find(input.userId))
await step.do('charge', async ({ step, span }) => {
  span.setAttribute('user.plan', user.plan)
  await step.do('authorize', () => gateway.authorize(user)) // nested → 'charge|authorize'
  await step.do('capture', () => gateway.capture(user))
})

回调接收 { step, span }：嵌套的 step 名称会加上父级前缀（parent|child），以及该步骤的 OTel span。步骤名称是缓存键——请在一个任务中保持名称稳定且唯一，并将步骤重命名视为缓存结果失效。

step.wait(name, durationMs) 是持久化睡眠：首次进入时会保存 startedAt，因此重放时只会等待剩余时间，而不会再次等待完整时长。step.waitUntil(name, date) 的行为相同，但会等待到绝对时间。

请牢记以下三个特性：

• 步骤级别不支持重试。 step.do 不接受选项对象；抛出错误会使整个任务失败，重试在任务级别进行（maxAttempts）。
• 取消是协作式的。 worker 失去其声明权（锁被抢占、心跳丢失）时，会中止任务的信号：step.wait 会立即拒绝，step.do 会拒绝启动新工作——但已在运行的步骤不会中断。长时间运行的 CPU 密集型步骤会一直占用其执行时段，直到返回。
• 对步骤使用 Promise.all 不会取消其他步骤。 一个并行步骤抛出错误时，其他步骤会继续运行直到完成；任务会等它们结束后才失败（并记录警告）。对于失败后不应继续运行的步骤，请使用顺序执行。

由于重放会重新执行缓存步骤周围的函数体，因此步骤之外的所有内容都应保持纯粹——步骤之间的副作用会在每次重试时运行。

排序、分组与并发

组是一个串行通道：共享同一 groupId 的作业会按 FIFO 顺序逐个运行，某个作业会阻塞其所在组，直到运行完成。如果没有 groupId（或 getGroupId），每个作业都会分配一个随机 UUID 组，并完全并行运行。使用实体 ID（例如 userId、tenantId）作为组，以便按实体串行处理任务。

priority 是一个普通数字，范围为 0…2²¹-1，数值越大优先级越高，默认值为 0。它不是 'high' | 'normal' 枚举。

以下四种上限彼此独立，且都为可选项：

上限	设置位置	默认值
命名空间中所有工作流的总量	WorkflowNamespace({ concurrency })	不限
单个工作流的所有 worker 总量	queueOptions.concurrency	不限
单个组内	queueOptions.groupConcurrency	1
单个 worker 进程内	workerOptions.concurrency / work({ concurrency })	1

worker 并发数默认为 1，这通常会让人意外：仅调用 work() 时，每个进程一次只会处理一个作业。

重试

maxAttempts 默认值为 1，即不重试。它会在入队时按作业应用，因此，只有当生产者与 worker 共享这些选项时，在 workerOptions 中设置的值才会对生产者入队的作业生效；如果生产者与 worker 是独立进程，建议使用 jobOptions.maxAttempts 或每次运行的 opts.maxAttempts。

退避策略在 worker 端按工作流设置：workerOptions.backoff(attempt) 返回指定尝试（从 1 开始）之前等待的毫秒数，默认使用 expBackoff()——带完全抖动的指数退避，base 为 500 ms，factor 为 2，cap 为 30 s。backoff: () => 0 是标准的测试覆盖设置。

Cron 计划

await onboarding.upsertSchedule('nightly-digest', {
  pattern: '0 3 * * *',
  input: { userId: 'system', name: 'digest' },
  tz: 'Europe/Berlin',
})

upsertSchedule(scheduleId, opts) 是幂等的：重复插入同一 ID 会原位替换，因此可以在每次启动时安全调用。input 会在此处立即根据 schema 验证，因此错误的负载会在注册时失败，而不是凌晨 3 点才失败。选项包括：pattern（Croner 语法，仅支持 cron 模式，不支持间隔形式）、input、tz（IANA 时区，默认使用注册进程的本地时区）、priority、groupId（默认为 scheduleId，用于串行处理各次触发）以及 skipIfRunning（默认 true，运行缓慢时会跳过下一次触发，而不是不断积压）。错过的触发会合并为一次运行。

removeSchedule(scheduleId) 和 getSchedules() 补全了该接口；后者会为每个计划返回 { scheduleId, pattern, tz, nextRun, lastFireAt, lastJobId }。

可观测性

注册 OTel SDK 后，追踪会自动启用：生产者 span 会将上下文传播到负载中，因此 worker 的作业 span 和每个步骤 span 都会关联到将作业入队的调用方。run 和 step.do 上下文中的 span 是当前活动 span，可直接在其上设置属性。

队列深度指标需通过 workerOptions.metrics（{ meter, prefix }，可在命名空间上设置，以应用于所有工作流）选择启用，并在调用 work() 时注册。导出的仪表和完整的 worker 调优选项（lockMs、maxStalledCount、stalledInterval、safetyTimeout、promoteBatchSize、keepFailed、onError、onFailed）请参阅 reference/options.md。