cloudflare-workflows

用途：Cloudflare Workflows 为 Workers 提供持久化的多步骤执行能力，步骤会持久化状态、可在重启后继续、支持重试，并可休眠数天。

主要能力：通过 WorkflowEntrypoint 的 run(event, step) 编写逻辑；step.do(name, callback) 执行并持久化返回值，可传入重试与超时配置（默认 retries.limit 5、delay 10000、backoff exponential、timeout 10 分钟）；支持 step.sleep 相对休眠与 step.sleepUntil 定时休眠（最长 365 天）；step.waitForEvent 等待外部事件，默认超时 24 小时，事件可在流程到达等待点前缓冲投递。实例支持 create、createBatch、get、pause、resume、terminate、restart、status、sendEvent，并具有 queued、running、paused、waiting、complete、errored、terminated 等状态。使用 NonRetryableError 可阻止重试，可 try-catch 捕获后继续。

用法要点：在 wrangler.jsonc 配置 workflows 绑定（name、binding、class_name），用 npx wrangler deploy 部署。CLI 可管理实例，如 wrangler workflows instances list/describe/terminate 与 trigger。

限制与规则：步骤须粒度合理、命名确定、必须 await，返回值小于 1 MiB；不要修改 event.payload（不持久化），不要将状态存于步骤之外。免费与付费计划在 CPU 时间、并发实例、状态总量、保留期等方面存在差异，具体数值见官方资料。

[以下均为资料原文，保留原样]

WorkflowEntrypoint

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

WorkflowEvent

interface WorkflowEvent<T> {
  payload: Readonly<T>; // Immutable input params
  timestamp: Date; // Creation time
  instanceId: string; // Unique instance ID
}

WorkflowStep

interface WorkflowStep {
  do<T>(name: string, callback: () => Promise<T>): Promise<T>;
  do<T>(name: string, config: StepConfig, callback: () => Promise<T>): Promise<T>;
  sleep(name: string, duration: Duration): Promise<void>;
  sleepUntil(name: string, timestamp: Date | number): Promise<void>;
  waitForEvent<T>(name: string, options: WaitOptions): Promise<T>;
}

快速开始

// src/index.ts
import { WorkflowEntrypoint, WorkflowStep, WorkflowEvent } from "cloudflare:workers";

interface Env {
  MY_WORKFLOW: Workflow;
}

interface Params {
  userId: string;
  action: string;
}

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent, step: WorkflowStep) {
    const user = await step.do("fetch user", async () => {
      const resp = await fetch(`https://api.example.com/users/${event.payload.userId}`);
      return resp.json();
    });

    await step.sleep("wait before processing", "1 hour");

    const result = await step.do("process action", async () => {
      return { processed: true, user: user.id };
    });

    return result; // Available in instance.status().output
  }
}

import { NonRetryableError } from "cloudflare:workflows";

const userData = await step.do("fetch user", async () => {
  return await fetchUser(userId);
});

let userData;
await step.do("fetch user", async () => {
  userData = await fetchUser(userId); // Will be lost!
});

const approval = await step.waitForEvent<{ approved: boolean }>("wait for approval", {
  type: "user_approval",
  timeout: "7 days",
});

await instance.sendEvent({
  type: "user_approval",
  payload: { approved },
});

wrangler.jsonc

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

部署

npx wrangler deploy

提高 CPU 限额

{
  "limits": {
    "cpu_ms": 300000 // 5 minutes
  }
}

Wrangler 命令

wrangler workflows instances list my-workflow
wrangler workflows instances describe my-workflow --id <instance-id>
wrangler workflows instances terminate my-workflow --id <instance-id>
wrangler workflows trigger my-workflow --params '{"key": "value"}'
wrangler workflows delete my-workflow

限制表（数据来自资料原文）：

项目	Free	Paid
每步 CPU 时间	10 ms	30 sec（上限 5 min）
每步墙钟时间	不限	不限
每步状态	1 MiB	1 MiB
事件负载	1 MiB	1 MiB
每实例总状态	100 MB	1 GB
最长休眠	365 天	365 天
每 Workflow 最大步数	1024	1024
并发实例	25	10,000
实例创建速率	100/sec	100/sec
排队实例	100,000	1,000,000
每实例子请求	50/req	1000/req
实例 ID 长度	100 chars	100 chars
已完成实例保留	3 days	30 days

定价（基于 Workers Standard）：

指标	Free	Paid
请求	100K/天（共享）	10M/月，+$0.30/M
CPU 时间	10 ms/次	30M ms/月，+$0.02/M ms
存储	1 GB	1 GB，+$0.20/GB-mo

注意：处于 waiting 状态（休眠、等待事件）的实例不计入并发限制。存储按所有实例计算，删除实例可释放存储，几分钟内更新。

技能介绍