mcpField
C
cloudflare-workflows

cloudflare-workflows

github.com/jackspace云与 DevOps1636 次安装12 周前更新
代码仓库

关于 cloudflare-workflows

cloudflare-workflows 是用于构建 Cloudflare Workers 工作流的技能,介绍步骤执行、重试、等待、事件处理、触发和实例管理。

主要功能

  1. 01使用 WorkflowEntrypoint 定义工作流
  2. 02运行可配置重试的命名步骤
  3. 03按时长、日期或事件等待
  4. 04触发和管理工作流实例
  5. 05使用 Wrangler 查看实例

适用场景

  1. 01构建包含可重试步骤的流程
  2. 02让工作流等待到指定时间
  3. 03收到外部事件后继续流程
  4. 04查看或管理运行中的实例

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/jackspace/claudeskillz --skill cloudflare-workflows

安装后在 Claude Code 中输入/cloudflare-workflows确认已加载。

附带文件README.mdSKILL.json

详细介绍

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 超时
  • 已在开发环境中测试
  • 已测试重试行为
  • 已测试错误场景

相关文档


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

评论

支持 Markdown