cloudflare-workflows

Cloudflare Workflows

状态：已可用于生产 ✅ 最后更新：2025-10-22 依赖项：cloudflare-worker-base（用于 Worker 设置） 最新版本：[email protected]、@cloudflare/[email protected]

---

快速入门（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) {
    // Access params from event.payload
    const { userId, email } = event.payload;

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

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

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

    // Optional: return final state
    return { completed: true, userId };
  }
}

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

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

3. 配置 Wrangler

wrangler.jsonc：

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

4. 部署和测试

# Deploy workflow
npm run deploy

# Trigger workflow (visit in browser or curl)
curl https://my-workflow.<subdomain>.workers.dev/

# View workflow instances
npx wrangler workflows instances list my-workflow

# Check instance status
npx wrangler workflows instances describe my-workflow <instance-id>

---

WorkflowEntrypoint 类

扩展 WorkflowEntrypoint

每个工作流都必须扩展 WorkflowEntrypoint 并实现 run() 方法：

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

类型参数：

• Env - 环境绑定（KV、D1、R2 等）
• Params - 通过 event.payload 传入的工作流参数类型

run() 方法

async run(
  event: WorkflowEvent<Params>,
  step: WorkflowStep
): Promise<T | void>

参数：

• event - 包含工作流元数据和负载
• step - 提供步骤方法（do、sleep、sleepUntil、waitForEvent）

返回值：

• 可选的返回值（必须可序列化）
• 可通过 instance.status() 获取返回值

示例：

export class OrderWorkflow extends WorkflowEntrypoint<Env, OrderParams> {
  async run(event: WorkflowEvent<OrderParams>, step: WorkflowStep) {
    const { orderId, customerId } = event.payload;

    // Access bindings via this.env
    const order = await this.env.DB.prepare(
      'SELECT * FROM orders WHERE id = ?'
    ).bind(orderId).first();

    const result = await step.do('process payment', async () => {
      // Payment processing
      return { paid: true, amount: order.total };
    });

    // Return final state
    return {
      orderId,
      status: 'completed',
      paidAmount: result.amount
    };
  }
}

---

步骤方法

step.do() - 执行工作

step.do<T>(
  name: string,
  config?: WorkflowStepConfig,
  callback: () => Promise<T>
): Promise<T>

或者（配置为可选）：

step.do<T>(
  name: string,
  callback: () => Promise<T>
): Promise<T>

参数：

• name - 步骤名称（用于可观测性）
• config（可选）- 重试配置
• callback - 执行工作的异步函数

返回值：

• callback 返回的值（必须可序列化）

示例：

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

// Step with retry config
const result = await step.do(
  'call payment API',
  {
    retries: {
      limit: 10,
      delay: '10 seconds',
      backoff: 'exponential'
    },
    timeout: '5 minutes'
  },
  async () => {
    const response = await fetch('https://payment-api.example.com/charge', {
      method: 'POST',
      body: JSON.stringify({ amount: 100 })
    });
    return await response.json();
  }
);

重要提示 - 序列化：

• 返回值必须是可序列化的 JSON
• ✅ 允许：字符串、数字、布尔值、数组、对象、null
• ❌ 禁止：函数、Symbol、循环引用、undefined
• 如果返回值不可序列化，步骤会抛出错误

---

step.sleep() - 相对休眠

step.sleep(name: string, duration: WorkflowDuration): Promise<void>

参数：

• name - 步骤名称
• duration - 数字（毫秒）或人类可读的字符串

接受的单位：

• "second" / "seconds"
• "minute" / "minutes"
• "hour" / "hours"
• "day" / "days"
• "week" / "weeks"
• "month" / "months"
• "year" / "years"

示例：

// Sleep for 5 minutes
await step.sleep('wait 5 minutes', '5 minutes');

// Sleep for 1 hour
await step.sleep('hourly delay', '1 hour');

// Sleep for 2 days
await step.sleep('wait 2 days', '2 days');

// Sleep using milliseconds
await step.sleep('wait 30 seconds', 30000);

// Common pattern: schedule daily task
await step.do('send daily report', async () => {
  // Send report
});
await step.sleep('wait until tomorrow', '1 day');
// Workflow continues next day

优先级：

• 从休眠中恢复的工作流优先于新实例
• 确保较早启动的工作流在新工作流启动前完成

---

step.sleepUntil() - 休眠至指定日期

step.sleepUntil(
  name: string,
  timestamp: Date | number
): Promise<void>

参数：

• name - 步骤名称
• timestamp - Date 对象或 UNIX 时间戳（毫秒）

示例：

// Sleep until specific date
const launchDate = new Date('2025-12-25T00:00:00Z');
await step.sleepUntil('wait for launch', launchDate);

// Sleep until UNIX timestamp
const timestamp = Date.parse('24 Oct 2024 13:00:00 UTC');
await step.sleepUntil('wait until time', timestamp);

// Sleep until next Monday 9am UTC
const nextMonday = new Date();
nextMonday.setDate(nextMonday.getDate() + ((1 + 7 - nextMonday.getDay()) % 7 || 7));
nextMonday.setUTCHours(9, 0, 0, 0);
await step.sleepUntil('wait until Monday 9am', nextMonday);

// Schedule work at specific time
await step.do('prepare campaign', async () => {
  // Prepare marketing campaign
});

const campaignLaunch = new Date('2025-11-01T12:00:00Z');
await step.sleepUntil('wait for campaign launch', campaignLaunch);

await step.do('launch campaign', async () => {
  // Launch campaign
});

---

step.waitForEvent() - 等待外部事件

step.waitForEvent<T>(
  name: string,
  options: { type: string; timeout?: string | number }
): Promise<T>

参数：

• name - 步骤名称
• options.type - 要匹配的事件类型
• options.timeout（可选）- 最长等待时间（默认：24 小时）

返回值：

• 通过 instance.sendEvent() 发送的事件负载

示例：

export class PaymentWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // Create payment intent
    await step.do('create payment intent', async () => {
      // Call Stripe API
    });

    // Wait for webhook from Stripe (max 1 hour)
    const webhookData = await step.waitForEvent<StripeWebhook>(
      'wait for payment confirmation',
      { type: 'stripe-webhook', timeout: '1 hour' }
    );

    // Continue based on webhook
    if (webhookData.status === 'succeeded') {
      await step.do('fulfill order', async () => {
        // Fulfill order
      });
    } else {
      await step.do('handle failed payment', async () => {
        // Handle failure
      });
    }
  }
}

// Worker receives webhook and sends event to workflow
export default {
  async fetch(req: Request, env: Env): Promise<Response> {
    if (req.url.includes('/webhook/stripe')) {
      const webhookData = await req.json();

      // Get workflow instance by ID (stored when created)
      const instance = await env.PAYMENT_WORKFLOW.get(instanceId);

      // Send event to waiting workflow
      await instance.sendEvent({
        type: 'stripe-webhook',
        payload: webhookData
      });

      return new Response('OK');
    }
  }
};

超时行为：

• 超时后会抛出错误，工作流可以重试或失败
• 如果超时不应导致工作流失败，请使用 try-catch 包裹

try {
  const event = await step.waitForEvent('wait for user input', {
    type: 'user-submitted',
    timeout: '10 minutes'
  });
} catch (error) {
  // Timeout occurred - handle gracefully
  await step.do('send reminder', async () => {
    // Send reminder to user
  });
}

---

WorkflowStepConfig

为各个步骤配置重试行为：

interface WorkflowStepConfig {
  retries?: {
    limit: number;          // Max retry attempts (Infinity allowed)
    delay: string | number; // Delay between retries
    backoff?: 'constant' | 'linear' | 'exponential';
  };
  timeout?: string | number; // Max time per attempt
}

默认配置

如果未提供配置，Workflows 将使用：

{
  retries: {
    limit: 5,
    delay: 10000,      // 10 seconds
    backoff: 'exponential'
  },
  timeout: '10 minutes'
}

重试示例

固定退避（每次延迟相同）：

await step.do(
  'send email',
  {
    retries: {
      limit: 3,
      delay: '30 seconds',
      backoff: 'constant'  // Always wait 30 seconds
    }
  },
  async () => {
    // Send email
  }
);

线性退避（逐步增加延迟）：

await step.do(
  'poll API',
  {
    retries: {
      limit: 5,
      delay: '1 minute',
      backoff: 'linear'  // 1m, 2m, 3m, 4m, 5m
    }
  },
  async () => {
    // Poll API
  }
);

指数退避（大多数情况下推荐）：

await step.do(
  'call rate-limited API',
  {
    retries: {
      limit: 10,
      delay: '10 seconds',
      backoff: 'exponential'  // 10s, 20s, 40s, 80s, 160s, ...
    },
    timeout: '5 minutes'
  },
  async () => {
    // API call
  }
);

无限重试：

await step.do(
  'critical operation',
  {
    retries: {
      limit: Infinity,  // Retry forever
      delay: '1 minute',
      backoff: 'exponential'
    }
  },
  async () => {
    // Operation that must succeed eventually
  }
);

不重试：

await step.do(
  'non-idempotent operation',
  {
    retries: {
      limit: 0  // Fail immediately on error
    }
  },
  async () => {
    // One-time operation
  }
);

---

错误处理

NonRetryableError

强制工作流立即失败，不进行重试：

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

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    await step.do('validate input', async () => {
      if (!event.payload.userId) {
        throw new NonRetryableError('userId is required');
      }

      // Validate user exists
      const user = await this.env.DB.prepare(
        'SELECT * FROM users WHERE id = ?'
      ).bind(event.payload.userId).first();

      if (!user) {
        // Terminal error - retrying won't help
        throw new NonRetryableError('User not found');
      }

      return user;
    });
  }
}

何时使用 NonRetryableError：

• ✅ 身份验证/授权失败
• ✅ 不会改变的无效输入
• ✅ 资源不存在（404）
• ✅ 验证错误
• ❌ 网络故障（应重试）
• ❌ 速率限制（应使用退避策略重试）
• ❌ 临时服务中断（应重试）

---

捕获错误以继续工作流

通过捕获步骤错误，避免整个工作流失败：

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // Critical step - workflow fails if this fails
    await step.do('process payment', async () => {
      // Payment processing
    });

    // Optional step - workflow continues even if it fails
    try {
      await step.do('send confirmation email', async () => {
        // Email sending
      });
    } catch (error) {
      console.log(`Email failed: ${error.message}`);

      // Do cleanup or alternative action
      await step.do('log email failure', async () => {
        await this.env.DB.prepare(
          'INSERT INTO failed_emails (user_id, error) VALUES (?, ?)'
        ).bind(event.payload.userId, error.message).run();
      });
    }

    // Workflow continues
    await step.do('update order status', async () => {
      // Update status
    });
  }
}

模式：优雅降级：

// Try primary service, fall back to secondary
let result;

try {
  result = await step.do('call primary API', async () => {
    return await callPrimaryAPI();
  });
} catch (error) {
  console.log('Primary API failed, trying backup');

  result = await step.do('call backup API', async () => {
    return await callBackupAPI();
  });
}

---

触发工作流

在 Workers 中使用

在 wrangler.jsonc 中配置绑定：

{
  "name": "trigger-worker",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-22",
  "workflows": [
    {
      "name": "my-workflow",
      "binding": "MY_WORKFLOW",
      "class_name": "MyWorkflow",
      "script_name": "workflow-worker"  // If workflow is in different Worker
    }
  ]
}

从 Worker 触发：

type Env = {
  MY_WORKFLOW: Workflow;
};

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

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

获取实例状态

// Get instance by ID
const instance = await env.MY_WORKFLOW.get(instanceId);

// Get status
const status = await instance.status();

console.log(status);
// {
//   status: 'running' | 'complete' | 'errored' | 'queued' | 'unknown',
//   error: string | null,
//   output: any  // Return value from run() if complete
// }

向运行中的实例发送事件

// Get instance
const instance = await env.MY_WORKFLOW.get(instanceId);

// Send event (will be received by step.waitForEvent)
await instance.sendEvent({
  type: 'user-action',
  payload: { action: 'approved' }
});

暂停和恢复

// Pause instance
await instance.pause();

// Resume instance
await instance.resume();

// Terminate instance
await instance.terminate();

---

工作流模式

模式 1：长时间运行的进程

export class VideoProcessingWorkflow extends WorkflowEntrypoint<Env, VideoParams> {
  async run(event: WorkflowEvent<VideoParams>, step: WorkflowStep) {
    const { videoId } = event.payload;

    // Step 1: Upload to processing service
    const uploadResult = await step.do('upload video', async () => {
      const video = await this.env.MY_BUCKET.get(`videos/${videoId}`);
      const response = await fetch('https://processor.example.com/upload', {
        method: 'POST',
        body: video?.body
      });
      return await response.json();
    });

    // Step 2: Wait for processing (could take hours)
    await step.sleep('wait for initial processing', '10 minutes');

    // Step 3: Poll for completion
    let processed = false;
    let attempts = 0;

    while (!processed && attempts < 20) {
      const status = await step.do(`check status attempt ${attempts}`, async () => {
        const response = await fetch(
          `https://processor.example.com/status/${uploadResult.jobId}`
        );
        return await response.json();
      });

      if (status.complete) {
        processed = true;
      } else {
        attempts++;
        await step.sleep(`wait before retry ${attempts}`, '5 minutes');
      }
    }

    // Step 4: Download processed video
    await step.do('download processed video', async () => {
      const response = await fetch(uploadResult.downloadUrl);
      const processed = await response.blob();
      await this.env.MY_BUCKET.put(`processed/${videoId}`, processed);
    });

    return { videoId, status: 'complete' };
  }
}

---

模式 2：事件驱动的审批流程

export class ApprovalWorkflow extends WorkflowEntrypoint<Env, ApprovalParams> {
  async run(event: WorkflowEvent<ApprovalParams>, step: WorkflowStep) {
    const { requestId, requesterId } = event.payload;

    // Step 1: Create approval request
    await step.do('create approval request', async () => {
      await this.env.DB.prepare(
        'INSERT INTO approvals (id, requester_id, status) VALUES (?, ?, ?)'
      ).bind(requestId, requesterId, 'pending').run();
    });

    // Step 2: Send notification to approvers
    await step.do('notify approvers', async () => {
      await sendNotification(requestId);
    });

    // Step 3: Wait for approval (max 7 days)
    let approvalEvent;

    try {
      approvalEvent = await step.waitForEvent<ApprovalEvent>(
        'wait for approval decision',
        { type: 'approval-decision', timeout: '7 days' }
      );
    } catch (error) {
      // Timeout - auto-reject
      await step.do('auto-reject due to timeout', async () => {
        await this.env.DB.prepare(
          'UPDATE approvals SET status = ? WHERE id = ?'
        ).bind('rejected', requestId).run();
      });

      return { requestId, status: 'rejected', reason: 'timeout' };
    }

    // Step 4: Process decision
    await step.do('process approval decision', async () => {
      await this.env.DB.prepare(
        'UPDATE approvals SET status = ?, approver_id = ? WHERE id = ?'
      ).bind(approvalEvent.approved ? 'approved' : 'rejected', approvalEvent.approverId, requestId).run();
    });

    // Step 5: Execute approved action (if approved)
    if (approvalEvent.approved) {
      await step.do('execute approved action', async () => {
        // Execute the action
      });
    }

    return { requestId, status: approvalEvent.approved ? 'approved' : 'rejected' };
  }
}

---

模式 3：定时工作流

export class DailyReportWorkflow extends WorkflowEntrypoint<Env, ReportParams> {
  async run(event: WorkflowEvent<ReportParams>, step: WorkflowStep) {
    // Calculate next 9am UTC
    const now = new Date();
    const tomorrow9am = new Date();
    tomorrow9am.setUTCDate(tomorrow9am.getUTCDate() + 1);
    tomorrow9am.setUTCHours(9, 0, 0, 0);

    // Sleep until tomorrow 9am
    await step.sleepUntil('wait until 9am tomorrow', tomorrow9am);

    // Generate report
    const report = await step.do('generate daily report', async () => {
      const results = await this.env.DB.prepare(
        'SELECT * FROM metrics WHERE date = ?'
      ).bind(now.toISOString().split('T')[0]).all();

      return {
        date: now.toISOString().split('T')[0],
        metrics: results.results
      };
    });

    // Send report
    await step.do('send report', async () => {
      await sendEmail({
        to: event.payload.recipients,
        subject: `Daily Report - ${report.date}`,
        body: formatReport(report.metrics)
      });
    });

    return { sent: true, date: report.date };
  }
}

---

模式 4：工作流串联

export class OrderWorkflow extends WorkflowEntrypoint<Env, OrderParams> {
  async run(event: WorkflowEvent<OrderParams>, step: WorkflowStep) {
    const { orderId } = event.payload;

    // Step 1: Process payment
    const paymentResult = await step.do('process payment', async () => {
      return await processPayment(orderId);
    });

    // Step 2: Trigger fulfillment workflow
    const fulfillmentInstance = await step.do('start fulfillment', async () => {
      return await this.env.FULFILLMENT_WORKFLOW.create({
        params: {
          orderId,
          paymentId: paymentResult.id
        }
      });
    });

    // Step 3: Wait for fulfillment to complete
    await step.sleep('wait for fulfillment', '5 minutes');

    // Step 4: Check fulfillment status
    const fulfillmentStatus = await step.do('check fulfillment', async () => {
      const instance = await this.env.FULFILLMENT_WORKFLOW.get(fulfillmentInstance.id);
      return await instance.status();
    });

    if (fulfillmentStatus.status === 'complete') {
      // Step 5: Send confirmation
      await step.do('send order confirmation', async () => {
        await sendConfirmation(orderId);
      });
    }

    return { orderId, status: 'complete' };
  }
}

---

Wrangler 命令

列出工作流实例

# List all instances of a workflow
npx wrangler workflows instances list my-workflow

# Filter by status
npx wrangler workflows instances list my-workflow --status running
npx wrangler workflows instances list my-workflow --status complete
npx wrangler workflows instances list my-workflow --status errored

描述实例

# Get detailed info about specific instance
npx wrangler workflows instances describe my-workflow <instance-id>

# Output shows:
# - Current status (running/complete/errored)
# - Each step with start/end times
# - Step outputs
# - Retry history
# - Any errors
# - Sleep state (if sleeping)

触发工作流（开发环境）

# Deploy workflow
npx wrangler deploy

# Trigger via HTTP (if Worker is set up to trigger)
curl https://my-workflow.<subdomain>.workers.dev/

---

状态持久化

可持久化的内容

工作流会自动持久化 step.do() 返回的状态：

✅ 可序列化类型：

• 基本类型：string、number、boolean、null
• 数组：[1, 2, 3]、['a', 'b', 'c']
• 对象：{ key: 'value' }、{ nested: { data: true } }
• 嵌套结构：{ users: [{ id: 1, name: 'Alice' }] }

❌ 不可序列化类型：

• 函数：() => {}
• 符号：Symbol('key')
• 循环引用：const obj = {}; obj.self = obj;
• undefined（请改用 null）
• 类实例（请序列化为普通对象）

示例——正确的序列化方式：

// ✅ Good - all values serializable
const result = await step.do('fetch data', async () => {
  return {
    users: [
      { id: 1, name: 'Alice', active: true },
      { id: 2, name: 'Bob', active: false }
    ],
    timestamp: Date.now(),
    metadata: null
  };
});

// ❌ Bad - contains function
const bad = await step.do('bad example', async () => {
  return {
    data: [1, 2, 3],
    transform: (x) => x * 2  // ❌ Function not serializable
  };
});
// This will throw an error!

跨步骤访问状态

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // Step 1: Get data
    const userData = await step.do('fetch user', async () => {
      return { id: 123, email: '[email protected]' };
    });

    // Step 2: Use data from step 1
    const orderData = await step.do('create order', async () => {
      return {
        userId: userData.id,      // ✅ Access previous step's data
        userEmail: userData.email,
        orderId: 'ORD-456'
      };
    });

    // Step 3: Use data from step 1 and 2
    await step.do('send confirmation', async () => {
      await sendEmail({
        to: userData.email,       // ✅ Still accessible
        subject: `Order ${orderData.orderId} confirmed`
      });
    });
  }
}

---

可观测性

内置指标

工作流会自动跟踪：

• 实例状态：排队中、运行中、已完成、出错、已暂停
• 步骤执行情况：开始／结束时间、持续时间、成功／失败
• 重试历史记录：尝试次数、错误、延迟
• 休眠状态：工作流的唤醒时间
• 输出：步骤和 run() 的返回值

在控制面板中查看指标

通过 Cloudflare 控制面板访问：

1. Workers & Pages
2. 选择工作流
3. 查看实例和指标

指标包括：

• 创建的实例总数
• 成功率／错误率
• 平均执行时间
• 步骤级别的性能

通过程序访问

// Get instance status
const instance = await env.MY_WORKFLOW.get(instanceId);
const status = await instance.status();

console.log(status);
// {
//   status: 'complete',
//   error: null,
//   output: { userId: '123', status: 'processed' }
// }

---

限制

功能	限制
工作流最长持续时间	30 天
每个工作流的步骤数上限	10,000
sleep/sleepUntil 最长持续时间	30 天
步骤超时时间上限	15 分钟
并发实例数上限	无限制（自动扩缩容）
负载大小上限	128 KB
步骤输出大小上限	128 KB
waitForEvent 超时时间上限	30 天
重试次数上限	Infinity（可配置）

注意：

• step.sleep() 和 step.sleepUntil() 不计入 10,000 步的上限
• 工作流的总运行时间最长可达 30 天
• 每个步骤的执行时间最长为 15 分钟
• 重试计为单独的尝试，而非单独的步骤

---

定价

需要 Workers Paid 计划（$5/月）

工作流执行：

• 每月前 10,000,000 次步骤执行：免费
• 超出后：每百万次步骤执行 $0.30

计为步骤执行的项目：

• 每次 step.do() 调用
• 每次步骤重试
• step.sleep()、step.sleepUntil()、step.waitForEvent() 不计入

费用示例：

• 含 5 个步骤且没有重试的工作流：5 次步骤执行
• 含 3 个步骤，其中 1 个步骤重试 2 次的工作流：5 次步骤执行（3 + 2）
• 每月运行 10M 个简单工作流（每个含 5 个步骤）：((50M - 10M) / 1M) × $0.30 = $12/月

---

务必做到 ✅

1. 使用描述性步骤名称——例如“获取用户数据”，而不是“步骤 1”
2. 只返回可序列化的值——基本类型、数组、普通对象
3. 对终止性错误使用 NonRetryableError——例如身份验证失败、输入无效
4. 配置重试次数上限——除非确有必要，否则避免无限重试
5. 对可选步骤捕获错误——如果步骤可以优雅失败，请使用 try-catch
6. 重试时使用指数退避——大多数情况下使用默认退避策略
7. 尽早验证输入——如果输入无效，使用 NonRetryableError 快速失败
8. 保存工作流实例 ID——存入 DB/KV，以便之后查询状态
9. 在人工介入环节使用 waitForEvent——例如审批、外部确认
10. 监控工作流指标——跟踪成功率和错误

---

绝不执行 ❌

1. 绝不从步骤中返回函数 - 会导致序列化错误
2. 绝不创建循环引用 - 会导致序列化失败
3. 绝不假设步骤会立即执行 - 步骤可能会重试或休眠
4. 绝不使用阻塞操作 - 异步工作请使用 step.do()
5. 绝不让载荷/输出超过 128 KB - 会导致失败
6. 绝不无限重试非幂等操作 - 请设置重试次数限制
7. 绝不忽略序列化错误 - 请修复数据结构
8. 绝不将工作流用于实时操作 - 请改用 Durable Objects
9. 绝不跳过关键步骤的错误处理 - 请使用 try-catch 包装，或使用 NonRetryableError
10. 绝不假设重试期间步骤顺序有保证 - 每个步骤都是独立的

---

故障排除

问题："Cannot perform I/O on behalf of a different request"

原因： 尝试在另一个请求处理程序中使用在某个请求上下文中创建的 I/O 对象

解决方案： 始终在 step.do() 回调中执行 I/O

// ❌ Bad - I/O outside step
const response = await fetch('https://api.example.com/data');
const data = await response.json();

await step.do('use data', async () => {
  // Using data from outside step's I/O context
  return data;  // This will fail!
});

// ✅ Good - I/O inside step
const data = await step.do('fetch data', async () => {
  const response = await fetch('https://api.example.com/data');
  return await response.json();  // ✅ Correct
});

---

问题：NonRetryableError 在开发环境和生产环境中的行为不同

已知问题： 在开发模式下抛出消息为空的 NonRetryableError 会导致重试，但在生产环境中可以正常工作

解决方法： 始终为 NonRetryableError 提供消息

// ❌ May retry in dev
throw new NonRetryableError();

// ✅ Works consistently
throw new NonRetryableError('User not found');

来源： workers-sdk#10113

---

问题："The requested module 'cloudflare:workers' does not provide an export named 'WorkflowEvent'"

原因： 导入错误，或 @cloudflare/workers-types 版本过旧

解决方案：

# Update types
npm install -D @cloudflare/workers-types@latest

# Ensure correct import
import { WorkflowEntrypoint, WorkflowStep, WorkflowEvent } from 'cloudflare:workers';
import { NonRetryableError } from 'cloudflare:workflows';

---

问题：步骤返回 undefined，而不是预期值

原因： 步骤回调没有返回值

解决方案： 始终在步骤回调中返回值

// ❌ Bad - no return
const result = await step.do('get data', async () => {
  const data = await fetchData();
  // Missing return!
});
console.log(result);  // undefined

// ✅ Good - explicit return
const result = await step.do('get data', async () => {
  const data = await fetchData();
  return data;  // ✅
});

---

问题：工作流实例卡在 "running" 状态

可能原因：

1. 步骤休眠时间过长
2. 步骤正在等待永远不会到来的事件
3. 步骤正在以较长的退避间隔重试

解决方案：

# Check instance details
npx wrangler workflows instances describe my-workflow <instance-id>

# Look for:
# - Sleep state (will show wake time)
# - Waiting for event (will show event type and timeout)
# - Retry history (will show attempts and delays)

---

生产环境检查清单

将工作流部署到生产环境之前：

• 为所有步骤设置描述性名称
• 为所有步骤配置重试次数限制
• 对终止性错误使用 NonRetryableError
• 为关键步骤设置错误处理
• 使用 try-catch 包装可选步骤
• 不返回不可序列化的值
• 载荷大小低于 128 KB
• 工作流持续时间不超过 30 天
• 存储实例 ID，以便查询状态
• 配置监控和告警
• 配置 waitForEvent 超时
• 已在开发环境中测试
• 已测试重试行为
• 已测试错误场景

---

相关文档

• Cloudflare 工作流文档
• 入门指南
• Workers API
• 休眠与重试
• 事件与参数
• 限制
• 定价
• 更新日志

---

最后更新： 2025-10-22 版本： 1.0.0 维护者： Jeremy Dawes | [email protected]