跳转到内容
搜索文档
↵选择
↑↓导航
ESC关闭

休眠与重试

最后更新 查看 MarkdownAgent 设置

本指南介绍如何让 Workflow 休眠和/或为 Workflow 步骤配置重试。

让 Workflow 休眠

你可以将 Workflow 休眠设为显式步骤，这在希望 Workflow 等待、提前调度工作或暂停直到输入或其他外部状态就绪时很有用。

休眠相对时长

使用 step.sleep 让 Workflow 休眠相对时长：

await step.sleep("sleep for a bit", "1 hour");

step.sleep 的第二个参数接受 number（毫秒）或人类可读格式，例如 "1 minute" 或 "26 hours"。以这种方式使用时，step.sleep 接受的单位为：

| "second"
| "minute"
| "hour"
| "day"
| "week"
| "month"
| "year"

休眠至固定日期

使用 step.sleepUntil 让 Workflow 休眠至特定 Date：当你从其他系统获得时间戳或希望「调度」工作在特定时间（例如 UTC 周日 9:00）执行时很有用。

// sleepUntil 的第二个参数接受 Date 对象
const workflowsLaunchDate = Date.parse("24 Oct 2024 13:00:00 UTC");
await step.sleepUntil("sleep until X times out", workflowsLaunchDate);

你也可以直接向 sleepUntil 提供 UNIX 时间戳（自 UNIX 纪元以来的毫秒数）。

重试步骤

Workflow 中对 step.do 的每次调用都接受可选的 StepConfig，允许你定义该步骤的重试行为。

如果不提供自己的重试配置，Workflows 将应用以下默认值：

const defaultConfig: WorkflowStepConfig = {
	retries: {
		limit: 5,
		delay: 10000,
		backoff: "exponential",
	},
	timeout: "10 minutes",
};

提供自己的 StepConfig 时，你可以配置：

  • 步骤的总尝试次数（每个步骤最多 10,000 次重试）
  • 尝试之间的延迟。使用固定时长（number 毫秒或人类可读字符串），或使用返回下一次延迟的函数。
  • 每次尝试之间应用的退避算法：constant、linear 或 exponential
  • 超时时间（时长），超过后将步骤视为失败（包括重试尝试期间，因为超时是按每次尝试设置的）

例如，要将步骤限制为 10 次重试，并在每次尝试之间应用指数延迟（从 10 秒开始），你可以将以下配置作为可选对象传递给 step.do：

let someState = await step.do(
	"call an API",
	{
		retries: {
			limit: 10, // 总尝试次数
			delay: "10 seconds", // 每次重试之间的延迟
			backoff: "exponential", // "constant" | "linear" | "exponential"
		},
		timeout: "30 minutes",
	},
	async () => {
		/* 步骤代码 */
	},
);

设置动态重试延迟

当下一次重试延迟应取决于失败尝试或抛出的错误时，使用延迟函数。这比使用 constant、linear 或 exponential 退避的固定延迟提供更多控制。适用于速率限制、下游提供商恢复和短暂网络故障。

延迟函数接收包含以下内容的参数：

返回时长字符串、毫秒数，或解析为其中任一值的 Promise。

await step.do(
	"sync customer",
	{
		retries: {
			limit: 5,
			delay: ({ ctx, error }) => {
				if (error.message.includes("rate limit")) {
					return `${ctx.attempt * 30} seconds`;
				}

				return "10 seconds";
			},
		},
	},
	async () => {
		await syncCustomer();
	},
);
await step.do(
	"sync customer",
	{
		retries: {
			limit: 5,
			delay: ({ ctx, error }) => {
				if (error.message.includes("rate limit")) {
					return `${ctx.attempt * 30} seconds`;
				}

				return "10 seconds";
			},
		},
	},
	async () => {
		await syncCustomer();
	},
);

强制 Workflow 实例失败

你也可以在步骤内抛出 NonRetryableError，强制 Workflow 实例失败且不重试。

当你检测到来自上游系统的终端（永久）错误（例如身份验证失败）或其他重试无帮助的错误时，这很有用。

// 导入 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("some step", async () => {
			if (!event.payload.data) {
				throw new NonRetryableError(
					"event.payload.data did not contain the expected payload",
				);
			}
		});
	}
}

Workflow 实例将立即失败，不会调用后续步骤，Workflow 也不会重试。

如果较早的步骤注册了回滚处理程序，这些处理程序仍会在实例进入终端状态之前运行。

注册回滚处理程序

你可以为 step.do() 附加回滚处理程序以实现 saga 式补偿。当 Workflow 后续失败时，Workflows 会按 step-start 的逆序运行已注册的回滚处理程序。

注册了回滚选项的失败步骤也可以与任何已注册回滚处理程序的已完成步骤一起参与回滚。例如，如果步骤在注册回滚后抛出 NonRetryableError，其回滚处理程序将运行且 output 设为 undefined。

import { WorkflowEntrypoint } from "cloudflare:workers";
import { NonRetryableError } from "cloudflare:workflows";

export class OrderWorkflow extends WorkflowEntrypoint {
	async run(_event, step) {
		await step.do(
			"reserve inventory",
			async () => {
				const reservation = await reserveInventory();
				return { reservationId: reservation.id };
			},
			{
				rollback: async ({ output }) => {
					const { reservationId } = output;
					await releaseInventory(reservationId);
				},
				rollbackConfig: {
					retries: { limit: 3, delay: "10 seconds", backoff: "linear" },
					timeout: "2 minutes",
				},
			},
		);

		await step.do("charge card", async () => {
			throw new NonRetryableError("payment processor rejected the charge");
		});
	}
}
import {
	WorkflowEntrypoint,
	type WorkflowEvent,
	type WorkflowStep,
} from "cloudflare:workers";
import { NonRetryableError } from "cloudflare:workflows";

export class OrderWorkflow extends WorkflowEntrypoint<Env> {
	async run(_event: WorkflowEvent<unknown>, step: WorkflowStep) {
		await step.do(
			"reserve inventory",
			async () => {
				const reservation = await reserveInventory();
				return { reservationId: reservation.id };
			},
			{
				rollback: async ({ output }) => {
					const { reservationId } = output as { reservationId: string };
					await releaseInventory(reservationId);
				},
				rollbackConfig: {
					retries: { limit: 3, delay: "10 seconds", backoff: "linear" },
					timeout: "2 minutes",
				},
			},
		);

		await step.do("charge card", async () => {
			throw new NonRetryableError("payment processor rejected the charge");
		});
	}
}

回滚处理程序接收：

  • error - 导致 Workflow 失败的错误。
  • output - 前向步骤返回的值，或步骤在返回前失败时为 undefined

你可以使用 rollbackConfig 控制回滚处理程序的重试行为。从回滚处理程序抛出 NonRetryableError 可立即停止重试。

捕获 Workflow 错误

任何传播到顶层的未捕获异常，或任何达到重试限制的步骤，都会导致 Workflow 以 Errored 状态结束执行。

如果要避免这种情况，你可以捕获 step 抛出的异常。当你需要触发清理任务或具有触发额外步骤的条件逻辑时很有用。

要让 Workflow 继续执行，用 try...catch 块包围允许失败的预期步骤。

...
await step.do('task', async () => {
	// 待完成的工作
});

try {
    await step.do('non-retryable-task', async () => {
		// 不应重试的工作
        throw new NonRetryableError('oh no');
    });
} catch (e) {
    console.log(`Step failed: ${e.message}`);
    await step.do('clean-up-task', async () => {
      // 清理代码
    });
}

// Workflow 不会失败，将继续执行

await step.do('next-task', async() => {
	// 更多工作
});
...

这篇文档对您有帮助吗？