cloudflare-workflows

Cloudflare Workflows

状态：已可用于生产环境 ✅ | 最后验证时间：2025-12-27 | 版本：3.0.0

依赖项：cloudflare-worker-base（用于 Worker 设置）

目录：快速入门 • 命令 • Agent • 核心概念 • 关键规则 • 常见错误 • 常见模式 • 何时加载参考资料 • 限制

---

快速入门（10 分钟）

1. 创建工作流

使用 Cloudflare Workflows 入门模板：

npm create cloudflare@latest my-workflow -- --template cloudflare/workflows-starter --git --deploy false
cd my-workflow

你将获得：

• WorkflowEntrypoint 类模板
• 用于触发工作流的 Worker
• 完整的 wrangler.jsonc 配置

2. 基本工作流结构

src/index.ts：

import { WorkflowEntrypoint, WorkflowStep, WorkflowEvent } from 'cloudflare:workers';

type Env = {
  MY_WORKFLOW: Workflow;
};

type Params = {
  userId: string;
  email: string;
};

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    const { userId, email } = event.payload;

    // Step 1: Do work with automatic retries
    const result = await step.do('process user', async () => {
      return { processed: true, userId };
    });

    // Step 2: Wait before next step
    await step.sleep('wait 1 hour', '1 hour');

    // Step 3: Continue workflow
    await step.do('send email', async () => {
      return { sent: true, email };
    });

    return { completed: true, userId };
  }
}

// Worker to trigger workflow
export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    const instance = await env.MY_WORKFLOW.create({
      params: { userId: '123', email: '[email protected]' }
    });

    return Response.json({
      id: instance.id,
      status: await instance.status()
    });
  }
};

模板：完整示例请参见 templates/basic-workflow.ts

3. 配置 wrangler.jsonc

{
  "name": "my-workflow",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-22",
  "workflows": [
    {
      "binding": "MY_WORKFLOW",
      "name": "my-workflow",
      "class_name": "MyWorkflow"
    }
  ]
}

模板：参见 templates/wrangler-workflows-config.jsonc

4. 部署

npm run deploy

---

命令

用于工作流开发的交互式斜杠命令：

命令	说明	使用时机
/workflow-setup	新工作流项目的完整向导	启动新项目，需要完整设置
/workflow-create	快速搭建工作流类的脚手架	向现有项目添加工作流
/workflow-debug	使用错误模式进行交互式调试	排查工作流问题
/workflow-test	在本地和远程测试工作流	验证工作流行为

使用示例：

/workflow-setup   # Full guided setup wizard
/workflow-create  # Quick workflow scaffolding
/workflow-debug   # Debug workflow issues
/workflow-test    # Test workflow execution

---

Agent

用于处理复杂工作流任务的自主 Agent：

Agent	说明	触发语句
workflow-debugger	自动检测并修复配置／运行时错误	"debug workflow", "fix workflow errors"
workflow-optimizer	分析性能、成本和可靠性	"optimize workflow", "improve performance"
workflow-setup-assistant	自主搭建项目脚手架	"setup workflow", "create first workflow"

主要能力：

• 调试器：分 6 个阶段进行分析，自动修复 I/O 上下文、序列化和导出问题
• 优化器：成本分析、可靠性评分、可执行的建议
• 设置助手：项目检测、自动搭建脚手架、验证

---

脚本

位于 scripts/ 目录中的自动化脚本：

脚本	用途
validate-workflow-config.sh	验证 wrangler.jsonc 配置
test-workflow.sh	创建并测试工作流实例
benchmark-workflow.sh	测量性能和成本
generate-workflow.sh	根据模板搭建新工作流的脚手架
check-workflow-limits.sh	检查是否符合 Cloudflare 限制

用法：

./scripts/validate-workflow-config.sh           # Check config
./scripts/test-workflow.sh my-workflow          # Test workflow
./scripts/benchmark-workflow.sh my-workflow 10  # Benchmark 10 runs
./scripts/generate-workflow.sh MyWorkflow       # Generate scaffold
./scripts/check-workflow-limits.sh src/workflows/my-workflow.ts

---

核心概念

WorkflowEntrypoint

每个工作流都必须继承 WorkflowEntrypoint：

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // Workflow logic here
  }
}

要点：

• Env：环境绑定（KV、D1 等）
• Params：创建工作流实例时传入的类型化载荷
• event：包含 id、payload、timestamp
• step：用于持久化执行的方法

Step 方法

为确保持久性，所有工作流操作都必须在步骤中完成：

// step.do - Execute work with automatic retries
await step.do('step name', async () => {
  return { result: 'data' };
});

// step.sleep - Wait for duration
await step.sleep('wait', '1 hour');

// step.sleepUntil - Wait until timestamp
await step.sleepUntil('wait until', Date.now() + 3600000);

// step.waitForEvent - Wait for external event
const event = await step.waitForEvent('payment received', 'payment.completed', {
  timeout: '30 minutes'
});

重要：所有 I/O（fetch、KV、D1、R2）都必须在 step.do() 回调内部执行！

参考：所有模式请参见 references/workflow-patterns.md

---

关键规则

务必做到 ✅

✅ 在 step.do() 内执行所有 I/O - 持久性所必需
✅ 使用命名步骤 - 便于调试
✅ 从步骤返回可进行 JSON 序列化的数据 - 状态持久化所必需
✅ 使用 step.sleep() 进行延迟 - 不要使用 setTimeout()
✅ 显式处理错误 - 在步骤回调中使用 try/catch
✅ 对永久性故障使用 NonRetryableError - 停止重试

工作流模式：以下模式请参见 references/workflow-patterns.md：

• 顺序工作流
• 并行执行
• 事件驱动工作流
• 定时工作流
• 人机协同工作流

绝不执行 ❌

❌ 绝不在 step.do() 外执行 I/O - 会导致“I/O context”错误 ❌ 绝不使用 setTimeout() 或 setInterval() - 请改用 step.sleep() ❌ 绝不返回不可序列化的数据 - 函数、Promise 等会导致失败 ❌ 绝不硬编码超时时间 - 请使用工作流配置 ❌ 绝不忽略 NonRetryableError - 这表示发生了永久性故障

---

五大关键错误

错误 #1：I/O Context 错误 ⚠️

错误：

Cannot perform I/O on behalf of a different request

原因： 在 step.do() 回调之外执行 I/O

解决方案：

// ❌ WRONG
const data = await fetch('https://api.example.com');
await step.do('use data', async () => {
  return data; // Error!
});

// ✅ CORRECT
const data = await step.do('fetch data', async () => {
  const response = await fetch('https://api.example.com');
  return await response.json();
});

错误 #2：序列化错误

错误：

Cannot serialize workflow state

原因： 从步骤返回不可序列化为 JSON 的数据

解决方案：

// ❌ WRONG
await step.do('process', async () => {
  return { fn: () => {} }; // Functions not serializable
});

// ✅ CORRECT
await step.do('process', async () => {
  return { result: 'data' }; // JSON-serializable
});

错误 #3：未抛出 NonRetryableError

错误： 工作流在永久性故障时无限重试

解决方案：

import { NonRetryableError } from 'cloudflare:workers';

await step.do('validate', async () => {
  if (!isValid) {
    throw new NonRetryableError('Invalid input'); // Stop retries
  }
  return { valid: true };
});

错误 #4：未找到 WorkflowEvent

错误：

WorkflowEvent 'payment.completed' not found

原因： waitForEvent 与触发器中的事件名称不匹配

解决方案：

// Workflow waits for event
const event = await step.waitForEvent('wait payment', 'payment.completed', {
  timeout: '30 minutes'
});

// Trigger event with EXACT same name
await instance.trigger('payment.completed', { amount: 100 });

错误 #5：工作流执行失败

错误：

Workflow execution failed: Step timeout exceeded

原因： 步骤超过 CPU 时间上限（30 秒）

解决方案：

// ❌ WRONG
await step.do('long task', async () => {
  for (let i = 0; i < 1000000; i++) {
    // Long computation
  }
});

// ✅ CORRECT - Break into smaller steps
for (let i = 0; i < 100; i++) {
  await step.do(`batch ${i}`, async () => {
    // Process batch
  });
}

---

所有问题：请参阅 references/common-issues.md，了解完整文档

---

常见模式

顺序工作流

步骤按顺序执行的基础工作流。每个步骤完成后才会开始下一个步骤。

适用场景：订单处理、用户引导、数据管道

加载 templates/basic-workflow.ts 查看完整示例

定时工作流

使用 step.sleep() 或 step.sleepUntil() 在步骤之间添加时间延迟的工作流。

适用场景：提醒序列、定时任务、延迟通知

加载 templates/scheduled-workflow.ts 查看完整示例

事件驱动工作流

使用 step.waitForEvent() 等待外部事件。务必设置超时时间，并使用 NonRetryableError 进行处理：

const payment = await step.waitForEvent('wait payment', 'payment.completed', {
  timeout: '30 minutes'
});
if (!payment) throw new NonRetryableError('Payment timeout');

加载 templates/workflow-with-events.ts 查看完整示例

带重试的工作流

永久性故障（404）使用 NonRetryableError，暂时性故障（5xx）使用常规 Error：

const data = await step.do('fetch', async () => {
  const response = await fetch(url);
  if (!response.ok) {
    if (response.status === 404) throw new NonRetryableError('Not found');
    throw new Error('Temporary failure'); // Will retry
  }
  return await response.json();
});

加载 templates/workflow-with-retries.ts 查看包含重试配置的完整示例

---

触发工作流

从 Worker 触发：通过 env.MY_WORKFLOW.create() 创建实例，使用 instance.status() 获取状态，并通过 instance.trigger() 触发事件。

从 Cron 触发：使用 scheduled() 处理程序按计划创建工作流实例。

加载 templates/worker-trigger.ts 查看完整的 Worker 触发示例 加载 templates/scheduled-workflow.ts 查看完整的 Cron 触发示例

---

何时加载参考资料

references/common-issues.md：遇到 I/O 上下文、序列化、NonRetryableError、事件命名或超时错误；排查工作流故障。

references/workflow-patterns.md：构建复杂编排、审批工作流、幂等模式或熔断器模式。

references/wrangler-commands.md：需要用于管理工作流实例、调试卡住的工作流或监控生产环境的 CLI 命令。

references/production-checklist.md：准备部署、需要部署前验证、设置监控或错误处理。

references/limits-quotas.md：达到实例、步骤或负载限制；优化成本；设计高吞吐量工作流。

references/2025-features.md：使用事件系统、增强型重试、实例生命周期控制或最新的 Workflows 功能。

references/metrics-analytics.md：设置监控、自定义指标、外部日志集成或工作流仪表板。

references/troubleshooting.md：处理复杂调试场景、卡住的实例、系统化诊断或性能问题。

templates/：basic-workflow.ts（顺序执行）、scheduled-workflow.ts（延迟/休眠）、workflow-with-events.ts（waitForEvent）、workflow-with-retries.ts（自定义重试）、worker-trigger.ts（Worker 触发器）、wrangler-workflows-config.jsonc（Wrangler 配置）、parallel-execution-workflow.ts（批量并行处理）、circuit-breaker-workflow.ts（弹性外部调用）

---

Wrangler 命令

关键命令：wrangler workflows create、wrangler workflows instances list/describe/terminate、wrangler deploy

加载 references/wrangler-commands.md，获取完整的 CLI 参考，其中包含所有工作流管理命令、工作流监控和卡住实例的调试方法。

---

状态持久化

工作流会自动在步骤之间持久化状态。无需手动管理状态：

export class StatefulWorkflow extends WorkflowEntrypoint {
  async run(event, step) {
    // Step 1 result is automatically persisted
    const result1 = await step.do('step 1', async () => {
      return { data: 'value' };
    });

    // Even if workflow crashes here, step 1 won't re-run
    await step.sleep('wait', '1 hour');

    // Step 2 can use step 1's result (still available after sleep)
    await step.do('step 2', async () => {
      console.log(result1.data); // 'value' - persisted!
    });
  }
}

要点：

• 步骤结果会自动持久化
• 已完成的步骤绝不会重新运行（即使发生崩溃或重启）
• 在工作流的整个生命周期内均可访问状态

---

限制

资源	限制
步骤 CPU 时间	30 秒
工作流持续时间	30 天
步骤负载大小	128 KB
工作流负载大小	128 KB
每个工作流的步骤数	1,000
并发实例数	每个工作流 1,000 个
事件负载大小	128 KB

解决方法：

• 大型数据：存储在 KV/R2 中，在步骤中传递键
• CPU 时间较长：拆分为更小的步骤
• 步骤较多：考虑使用子工作流

定价

• 持续时间：每百万 GB-s $0.02（与 Workers 相同）
• 请求数：每百万次 $0.15（创建工作流 + 执行步骤）
• 状态存储：已包含（无额外费用）
• 休眠：免费（休眠期间不使用 CPU）

费用示例（1M 次工作流运行）：

• 每个工作流 5 个步骤 = 5M 次请求 = $0.75
• 每个步骤 10ms = 50GB-s = $0.001
• 总计：每百万个工作流约 $0.75

---

故障排查

“I/O context”错误

解决方法：将所有 I/O 移入 step.do() 回调 → 参见 references/common-issues.md #1

“Serialization error”

解决方法：步骤仅返回可序列化为 JSON 的数据 → 参见 references/common-issues.md #2

工作流无限重试

解决方法：对于永久性故障，抛出 NonRetryableError → 参见 references/common-issues.md #3

“WorkflowEvent not found”

解决方法：确保事件名称完全一致 → 参见 references/common-issues.md #4

“Step timeout exceeded”

解决方法：将耗时较长的计算拆分为更小的步骤 → 参见 references/common-issues.md #5

---

生产环境检查清单

部署前 10 项检查清单：I/O 上下文隔离、JSON 序列化、NonRetryableError 的使用、事件名称一致性、步骤时长限制、错误处理、重试配置、超时、工作流命名和监控。

加载 references/production-checklist.md，查看包含详细说明、代码示例、验证步骤和部署工作流的完整检查清单。

---

官方文档

工作流：https://developers.cloudflare.com/workflows/ • API 参考：https://developers.cloudflare.com/workflows/reference/ • 示例：https://developers.cloudflare.com/workflows/examples/ • 博客：https://blog.cloudflare.com/cloudflare-workflows/