← Cloudflare Workflows / workflows / build

Паузы и повторные попытки

В этом руководстве подробно описано, как поставить 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: это может быть полезно, если у вас есть отметка времени из другой системы или вы хотите «запланировать» выполнение работы на конкретное время (например, воскресенье, 9 утра UTC).

// sleepUntil accepts a Date object as its second argument
const workflowsLaunchDate = Date.parse("24 Oct 2024 13:00:00 UTC");
await step.sleepUntil("sleep until X times out", workflowsLaunchDate);

Вы также можете передать метку времени UNIX (в миллисекундах с начала эпохи UNIX) напрямую в sleepUntil.

Каждый вызов step.do в Workflow принимает необязательный StepConfig, которая позволяет определить поведение повторных попыток для этого шага.

Если вы не задаёте собственную конфигурацию повторных попыток, Workflows применяет следующие значения по умолчанию:

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

При использовании собственного StepConfig, можно настроить следующее:

• Общее количество попыток выполнения шага (не более 10,000 повторов на шаг)
• Задержка между попытками. Укажите фиксированную длительность как number в миллисекундах или строку в человекочитаемом формате, либо используйте функцию, которая возвращает следующую задержку.
• Какой алгоритм задержки (backoff) применять между попытками: один из constant, linear, или exponential
• Через какое время (в формате duration) шаг следует считать неудачным, включая повторные попытки, так как тайм-аут задаётся для каждой попытки отдельно

Например, чтобы ограничить шаг 10 повторными попытками и задать экспоненциальную задержку (начиная с 10 секунд) между попытками, передайте следующую конфигурацию в качестве необязательного объекта в step.do:

let someState = await step.do(
	"call an API",
	{
		retries: {
			limit: 10, // The total number of attempts
			delay: "10 seconds", // Delay between each retry
			backoff: "exponential", // Any of "constant" | "linear" | "exponential";
		},
		timeout: "30 minutes",
	},
	async () => {
		/* Step code goes here */
	},
);

Используйте функцию задержки, если следующая задержка перед повтором должна зависеть от неудачной попытки или выброшенной ошибки. Это даёт больше контроля, чем фиксированная задержка с constant, linear, или exponential задержка (backoff). Это полезно при ограничениях скорости, восстановлении после сбоев у сторонних провайдеров и кратковременных сетевых сбоях.

Функция задержки получает объект со следующими полями:

• ctx : текущий WorkflowStepContext, включая ctx.attempt.
• error : ошибка, из-за которой была запущена повторная попытка.

Верните строку длительности, число в миллисекундах или 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 изнутри шага.

Это может быть полезно, когда вы обнаруживаете финальную (постоянную) ошибку от вышестоящей системы, например ошибку аутентификации, или другие ошибки, при которых повторные попытки не помогут.

// Import the NonRetryableError definition
import {
	WorkflowEntrypoint,
	WorkflowStep,
	WorkflowEvent,
} from "cloudflare:workers";
import { NonRetryableError } from "cloudflare:workflows";

// In your step code:
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 : значение, возвращённое шагом forward, или undefined если шаг завершился ошибкой до возврата значения

Вы можете использовать rollbackConfig чтобы управлять поведением повторных попыток для обработчика отката. Выбросите NonRetryableError из обработчика отката, чтобы немедленно прекратить повторные попытки.

Любые необработанные исключения, которые доходят до верхнего уровня, а также шаги, достигшие лимита повторных попыток, приводят к завершению выполнения Workflow в состоянии Errored состояние.

Чтобы избежать этого, вы можете перехватывать исключения, генерируемые step. Это может быть полезно, если нужно запустить задачи очистки или использовать условную логику, запускающую дополнительные шаги.

Чтобы Workflow продолжил выполнение, оберните шаги, которым разрешено завершаться с ошибкой, в try...catch блок.

...
await step.do('task', async () => {
	// work to be done
});

try {
    await step.do('non-retryable-task', async () => {
		// work not to be retried
        throw new NonRetryableError('oh no');
    });
} catch (e) {
    console.log(`Step failed: ${e.message}`);
    await step.do('clean-up-task', async () => {
      // Clean up code here
    });
}

// the Workflow will not fail and will continue its execution

await step.do('next-task', async() => {
	// more work to be done
});
...