INTEGRITY Документация

← Cloudflare Workflows / workflows / build

Правила Workflows

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

Это небольшое руководство о том, как создавать более отказоустойчивые и корректные Workflow.

Убедитесь, что вызовы API/привязки идемпотентны

Поскольку шаг может выполняться повторно несколько раз, шаги должны (в идеале) быть идемпотентными. Идемпотентность означает, что операцию (в данном случае шаг) можно выполнять многократно, не меняя результат по сравнению с первым выполнением.

Предположим, что у вас есть Workflow, который списывает деньги с клиентов, и вы не хотите случайно списать их дважды. Перед списанием следует проверить, что оплата уже не была произведена:

index.js
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		const customer_id = 123456;
		// ✅ Good: Non-idempotent API/Binding calls are always done **after** checking if the operation is
		// still needed.
		await step.do(
			`charge ${customer_id} for its monthly subscription`,
			async () => {
				// API call to check if customer was already charged
				const subscription = await fetch(
					`https://payment.processor/subscriptions/${customer_id}`,
				).then((res) => res.json());

				// return early if the customer was already charged, this can happen if the destination service dies
				// in the middle of the request but still commits it, or if the Workflows Engine restarts.
				if (subscription.charged) {
					return;
				}

				// non-idempotent call, this operation can fail and retry but still commit in the payment
				// processor - which means that, on retry, it would mischarge the customer again if the above checks
				// were not in place.
				return await fetch(
					`https://payment.processor/subscriptions/${customer_id}`,
					{
						method: "POST",
						body: JSON.stringify({ amount: 10.0 }),
					},
				);
			},
		);
	}
}
index.ts
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		const customer_id = 123456;
		// ✅ Good: Non-idempotent API/Binding calls are always done **after** checking if the operation is
		// still needed.
		await step.do(
			`charge ${customer_id} for its monthly subscription`,
			async () => {
				// API call to check if customer was already charged
				const subscription = await fetch(
					`https://payment.processor/subscriptions/${customer_id}`,
				).then((res) => res.json());

				// return early if the customer was already charged, this can happen if the destination service dies
				// in the middle of the request but still commits it, or if the Workflows Engine restarts.
				if (subscription.charged) {
					return;
				}

				// non-idempotent call, this operation can fail and retry but still commit in the payment
				// processor - which means that, on retry, it would mischarge the customer again if the above checks
				// were not in place.
				return await fetch(
					`https://payment.processor/subscriptions/${customer_id}`,
					{
						method: "POST",
						body: JSON.stringify({ amount: 10.0 }),
					},
				);
			},
		);
	}
}

Разбивайте шаги на небольшие операции

Шаги должны быть как можно более самодостаточными. Это делает вашу логику более устойчивой к сбоям сторонних API, сетевым ошибкам и другим подобным ситуациям.

Также это можно рассматривать как транзакцию или единицу работы.

index.js
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// ✅ Good: Unrelated API/Binding calls are self-contained, so that in case one of them fails
		// it can retry them individually. It also has an extra advantage: you can control retry or
		// timeout policies for each granular step - you might not to want to overload http.cat in
		// case of it being down.
		const httpCat = await step.do("get cutest cat from KV", async () => {
			return await this.env.KV.get("cutest-http-cat");
		});

		const image = await step.do("fetch cat image from http.cat", async () => {
			return await fetch(`https://http.cat/${httpCat}`);
		});
	}
}
index.ts
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// ✅ Good: Unrelated API/Binding calls are self-contained, so that in case one of them fails
		// it can retry them individually. It also has an extra advantage: you can control retry or
		// timeout policies for each granular step - you might not to want to overload http.cat in
		// case of it being down.
		const httpCat = await step.do("get cutest cat from KV", async () => {
			return await this.env.KV.get("cutest-http-cat");
		});

		const image = await step.do("fetch cat image from http.cat", async () => {
			return await fetch(`https://http.cat/${httpCat}`);
		});
	}
}

Иначе весь Workflow может оказаться не таким надёжным, как вы предполагаете, и вы можете столкнуться с непредсказуемым поведением. Чтобы этого избежать, соблюдайте следующие правила:

index.js
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// 🔴 Bad: you are calling two separate services from within the same step. This might cause
		// some extra calls to the first service in case the second one fails, and in some cases, makes
		// the step non-idempotent altogether
		const image = await step.do("get cutest cat from KV", async () => {
			const httpCat = await this.env.KV.get("cutest-http-cat");
			return fetch(`https://http.cat/${httpCat}`);
		});
	}
}
index.ts
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// 🔴 Bad: you are calling two separate services from within the same step. This might cause
		// some extra calls to the first service in case the second one fails, and in some cases, makes
		// the step non-idempotent altogether
		const image = await step.do("get cutest cat from KV", async () => {
			const httpCat = await this.env.KV.get("cutest-http-cat");
			return fetch(`https://http.cat/${httpCat}`);
		});
	}
}

Не полагайтесь на состояние вне шага

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

Это означает, что не следует хранить состояние вне шага:

index.js
function getRandomInt(min, max) {
	const minCeiled = Math.ceil(min);
	const maxFloored = Math.floor(max);
	return Math.floor(Math.random() * (maxFloored - minCeiled) + minCeiled); // The maximum is exclusive and the minimum is inclusive
}

export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// 🔴 Bad: `imageList` will be not persisted across engine's lifetimes. Which means that after hibernation,
		// `imageList` will be empty again, even though the following two steps have already ran.
		const imageList = [];

		await step.do("get first cutest cat from KV", async () => {
			const httpCat = await this.env.KV.get("cutest-http-cat-1");

			imageList.push(httpCat);
		});

		await step.do("get second cutest cat from KV", async () => {
			const httpCat = await this.env.KV.get("cutest-http-cat-2");

			imageList.push(httpCat);
		});

		// A long sleep can (and probably will) hibernate the engine which means that the first engine lifetime ends here
		await step.sleep("💤💤💤💤", "3 hours");

		// When this runs, it will be on the second engine lifetime - which means `imageList` will be empty.
		await step.do(
			"choose a random cat from the list and download it",
			async () => {
				const randomCat = imageList.at(getRandomInt(0, imageList.length));
				// this will fail since `randomCat` is undefined because `imageList` is empty
				return await fetch(`https://http.cat/${randomCat}`);
			},
		);
	}
}
index.ts
function getRandomInt(min, max) {
	const minCeiled = Math.ceil(min);
	const maxFloored = Math.floor(max);
	return Math.floor(Math.random() * (maxFloored - minCeiled) + minCeiled); // The maximum is exclusive and the minimum is inclusive
}

export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// 🔴 Bad: `imageList` will be not persisted across engine's lifetimes. Which means that after hibernation,
		// `imageList` will be empty again, even though the following two steps have already ran.
		const imageList: string[] = [];

		await step.do("get first cutest cat from KV", async () => {
			const httpCat = await this.env.KV.get("cutest-http-cat-1");

			imageList.push(httpCat);
		});

		await step.do("get second cutest cat from KV", async () => {
			const httpCat = await this.env.KV.get("cutest-http-cat-2");

			imageList.push(httpCat);
		});

		// A long sleep can (and probably will) hibernate the engine which means that the first engine lifetime ends here
		await step.sleep("💤💤💤💤", "3 hours");

		// When this runs, it will be on the second engine lifetime - which means `imageList` will be empty.
		await step.do(
			"choose a random cat from the list and download it",
			async () => {
				const randomCat = imageList.at(getRandomInt(0, imageList.length));
				// this will fail since `randomCat` is undefined because `imageList` is empty
				return await fetch(`https://http.cat/${randomCat}`);
			},
		);
	}
}

Вместо этого верхнеуровневое состояние следует формировать исключительно из step.do возвращает:

index.js
function getRandomInt(min, max) {
	const minCeiled = Math.ceil(min);
	const maxFloored = Math.floor(max);
	return Math.floor(Math.random() * (maxFloored - minCeiled) + minCeiled); // The maximum is exclusive and the minimum is inclusive
}

export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// ✅ Good: imageList state is exclusively comprised of step returns - this means that in the event of
		// multiple engine lifetimes, imageList will be built accordingly
		const imageList = await Promise.all([
			step.do("get first cutest cat from KV", async () => {
				return await this.env.KV.get("cutest-http-cat-1");
			}),

			step.do("get second cutest cat from KV", async () => {
				return await this.env.KV.get("cutest-http-cat-2");
			}),
		]);

		// A long sleep can (and probably will) hibernate the engine which means that the first engine lifetime ends here
		await step.sleep("💤💤💤💤", "3 hours");

		// When this runs, it will be on the second engine lifetime - but this time, imageList will contain
		// the two most cutest cats
		await step.do(
			"choose a random cat from the list and download it",
			async () => {
				const randomCat = imageList.at(getRandomInt(0, imageList.length));
				// this will eventually succeed since `randomCat` is defined
				return await fetch(`https://http.cat/${randomCat}`);
			},
		);
	}
}
index.ts
function getRandomInt(min, max) {
	const minCeiled = Math.ceil(min);
	const maxFloored = Math.floor(max);
	return Math.floor(Math.random() * (maxFloored - minCeiled) + minCeiled); // The maximum is exclusive and the minimum is inclusive
}

export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// ✅ Good: imageList state is exclusively comprised of step returns - this means that in the event of
		// multiple engine lifetimes, imageList will be built accordingly
		const imageList: string[] = await Promise.all([
			step.do("get first cutest cat from KV", async () => {
				return await this.env.KV.get("cutest-http-cat-1");
			}),

			step.do("get second cutest cat from KV", async () => {
				return await this.env.KV.get("cutest-http-cat-2");
			}),
		]);

		// A long sleep can (and probably will) hibernate the engine which means that the first engine lifetime ends here
		await step.sleep("💤💤💤💤", "3 hours");

		// When this runs, it will be on the second engine lifetime - but this time, imageList will contain
		// the two most cutest cats
		await step.do(
			"choose a random cat from the list and download it",
			async () => {
				const randomCat = imageList.at(getRandomInt(0, imageList.length));
				// this will eventually succeed since `randomCat` is defined
				return await fetch(`https://http.cat/${randomCat}`);
			},
		);
	}
}

Избегайте побочных эффектов вне step.do

Не рекомендуется писать код с побочными эффектами вне шагов, если только вы не хотите, чтобы он выполнялся повторно: движок Workflow может перезапуститься во время работы экземпляра. При перезапуске логика шагов сохранится, а логика вне шагов может выполниться повторно.

Например, console.log() вне шагов workflow может привести к тому, что журналы будут выводиться дважды при перезапуске движка.

Однако логику, использующую несериализуемые ресурсы, например подключение к базе данных, следует выполнять вне шагов. Операции за пределами step.do может повторяться более одного раза из-за особенностей жизненного цикла инстансов Workflows.

index.js
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// 🔴 Bad: creating instances outside of steps
		// This might get called more than once creating more instances than expected
		const badInstance = await this.env.ANOTHER_WORKFLOW.create();

		// 🔴 Bad: using non-deterministic functions outside of steps
		// this will produce different results if the instance has to restart, different runs of the same instance
		// might go through different paths
		const badRandom = Math.random();

		if (badRandom > 0) {
			// do some stuff
		}

		// ⚠️ Warning: This log may happen many times
		console.log("This might be logged more than once");

		await step.do("do some stuff and have a log for when it runs", async () => {
			// do some stuff

			// this log will only appear once
			console.log("successfully did stuff");
		});

		// ✅ Good: wrap non-deterministic function in a step
		// after running successfully will not run again
		const goodRandom = await step.do("create a random number", async () => {
			return Math.random();
		});

		// ✅ Good: calls that have no side effects can be done outside of steps
		// For Hyperdrive, create the connection inside each step instead of here.
		const db = createDBConnection(this.env.DB_URL, this.env.DB_TOKEN);

		// ✅ Good: run functions with side effects inside of a step
		// after running successfully will not run again
		const goodInstance = await step.do(
			"good step that returns state",
			async () => {
				const instance = await this.env.ANOTHER_WORKFLOW.create();

				return instance;
			},
		);
	}
}
index.ts
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// 🔴 Bad: creating instances outside of steps
		// This might get called more than once creating more instances than expected
		const badInstance = await this.env.ANOTHER_WORKFLOW.create();

		// 🔴 Bad: using non-deterministic functions outside of steps
		// this will produce different results if the instance has to restart, different runs of the same instance
		// might go through different paths
		const badRandom = Math.random();

		if (badRandom > 0) {
			// do some stuff
		}

		// ⚠️ Warning: This log may happen many times
		console.log("This might be logged more than once");

		await step.do("do some stuff and have a log for when it runs", async () => {
			// do some stuff

			// this log will only appear once
			console.log("successfully did stuff");
		});

		// ✅ Good: wrap non-deterministic function in a step
		// after running successfully will not run again
		const goodRandom = await step.do("create a random number", async () => {
			return Math.random();
		});

		// ✅ Good: calls that have no side effects can be done outside of steps
		// For Hyperdrive, create the connection inside each step instead of here.
		const db = createDBConnection(this.env.DB_URL, this.env.DB_TOKEN);

		// ✅ Good: run functions with side effects inside of a step
		// after running successfully will not run again
		const goodInstance = await step.do(
			"good step that returns state",
			async () => {
				const instance = await this.env.ANOTHER_WORKFLOW.create();

				return instance;
			},
		);
	}
}

Не изменяйте входящие события

event переданный в run метод неизменяем: изменения, внесённые в событие, не сохраняются между шагами и/или перезапусками Workflow.

index.js
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// 🔴 Bad: Mutating the event
		// This will not be persisted across steps and `event.payload` will
		// take on its original value.
		await step.do("bad step that mutates the incoming event", async () => {
			let userData = await this.env.KV.get(event.payload.user);
			event.payload = userData;
		});

		// ✅ Good: persist data by returning it as state from your step
		// Use that state in subsequent steps
		let userData = await step.do("good step that returns state", async () => {
			return await this.env.KV.get(event.payload.user);
		});

		let someOtherData = await step.do(
			"following step that uses that state",
			async () => {
				// Access to userData here
				// Will always be the same if this step is retried
			},
		);
	}
}
index.ts
interface MyEvent {
	user: string;
	data: string;
}

export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<MyEvent>, step: WorkflowStep) {
		// 🔴 Bad: Mutating the event
		// This will not be persisted across steps and `event.payload` will
		// take on its original value.
		await step.do("bad step that mutates the incoming event", async () => {
			let userData = await this.env.KV.get(event.payload.user);
			event.payload = userData;
		});

		// ✅ Good: persist data by returning it as state from your step
		// Use that state in subsequent steps
		let userData = await step.do("good step that returns state", async () => {
			return await this.env.KV.get(event.payload.user);
		});

		let someOtherData = await step.do(
			"following step that uses that state",
			async () => {
				// Access to userData here
				// Will always be the same if this step is retried
			},
		);
	}
}

Присваивайте шагам детерминированные имена

Названия шагов должны быть детерминированными, то есть не зависеть от текущей даты и времени, случайных значений и подобных факторов. Это обеспечивает кэширование их состояния и не позволяет шагу выполняться повторно без необходимости. В вашем Workflow название шага играет роль «ключа кэша».

index.js
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// 🔴 Bad: Naming the step non-deterministically prevents it from being cached
		// This will cause the step to be re-run if subsequent steps fail.
		await step.do(`step #1 running at: ${Date.now()}`, async () => {
			let userData = await this.env.KV.get(event.payload.user);
			// Do not mutate event.payload
			event.payload = userData;
		});

		// ✅ Good: give steps a deterministic name.
		// Return dynamic values in your state, or log them instead.
		let state = await step.do("fetch user data from KV", async () => {
			let userData = await this.env.KV.get(event.payload.user);
			console.log(`fetched at ${Date.now()}`);
			return userData;
		});

		// ✅ Good: steps that are dynamically named are constructed in a deterministic way.
		// In this case, `catList` is a step output, which is stable, and `catList` is
		// traversed in a deterministic fashion (no shuffles or random accesses) so,
		// it's fine to dynamically name steps (e.g: create a step per list entry).
		let catList = await step.do("get cat list from KV", async () => {
			return await this.env.KV.get("cat-list");
		});

		for (const cat of catList) {
			await step.do(`get cat: ${cat}`, async () => {
				return await this.env.KV.get(cat);
			});
		}
	}
}
index.ts
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// 🔴 Bad: Naming the step non-deterministically prevents it from being cached
		// This will cause the step to be re-run if subsequent steps fail.
		await step.do(`step #1 running at: ${Date.now()}`, async () => {
			let userData = await this.env.KV.get(event.payload.user);
			// Do not mutate event.payload
			event.payload = userData;
		});

		// ✅ Good: give steps a deterministic name.
		// Return dynamic values in your state, or log them instead.
		let state = await step.do("fetch user data from KV", async () => {
			let userData = await this.env.KV.get(event.payload.user);
			console.log(`fetched at ${Date.now()}`);
			return userData;
		});

		// ✅ Good: steps that are dynamically named are constructed in a deterministic way.
		// In this case, `catList` is a step output, which is stable, and `catList` is
		// traversed in a deterministic fashion (no shuffles or random accesses) so,
		// it's fine to dynamically name steps (e.g: create a step per list entry).
		let catList = await step.do("get cat list from KV", async () => {
			return await this.env.KV.get("cat-list");
		});

		for (const cat of catList) {
			await step.do(`get cat: ${cat}`, async () => {
				return await this.env.KV.get(cat);
			});
		}
	}
}

Будьте внимательны с Promise.race() и Promise.any()

Workflows позволяет использовать шаги в рамках Promise.race() или Promise.any() методы как способ достичь параллельного выполнения шагов. Однако следует учитывать ряд нюансов.

Из-за особенностей жизненного цикла экземпляров Workflows, а также того, что шаг внутри Promise выполняется до самого завершения, шаг, возвращаемый при первом проходе, может не быть тем самым кэшированным шагом, поскольку шаги кэшируются по своим именам.

index.js
// helper sleep method
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// 🔴 Bad: The `Promise.race` is not surrounded by a `step.do`, which may cause undeterministic caching behavior.
		const race_return = await Promise.race([
			step.do("Promise first race", async () => {
				await sleep(1000);
				return "first";
			}),
			step.do("Promise second race", async () => {
				return "second";
			}),
		]);

		await step.sleep("Sleep step", "2 hours");

		return await step.do("Another step", async () => {
			// This step will return `first`, even though the `Promise.race` first returned `second`.
			return race_return;
		});
	}
}
index.ts
// helper sleep method
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// 🔴 Bad: The `Promise.race` is not surrounded by a `step.do`, which may cause undeterministic caching behavior.
		const race_return = await Promise.race([
			step.do("Promise first race", async () => {
				await sleep(1000);
				return "first";
			}),
			step.do("Promise second race", async () => {
				return "second";
			}),
		]);

		await step.sleep("Sleep step", "2 hours");

		return await step.do("Another step", async () => {
			// This step will return `first`, even though the `Promise.race` first returned `second`.
			return race_return;
		});
	}
}

Для обеспечения согласованности рекомендуем обернуть Promise.race() или Promise.any() внутри step.do(), так как это обеспечит согласованность кеширования при нескольких проходах.

index.js
// helper sleep method
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// ✅ Good: The `Promise.race` is surrounded by a `step.do`, ensuring deterministic caching behavior.
		const race_return = await step.do("Promise step", async () => {
			return await Promise.race([
				step.do("Promise first race", async () => {
					await sleep(1000);
					return "first";
				}),
				step.do("Promise second race", async () => {
					return "second";
				}),
			]);
		});

		await step.sleep("Sleep step", "2 hours");

		return await step.do("Another step", async () => {
			// This step will return `second` because the `Promise.race` was surround by the `step.do` method.
			return race_return;
		});
	}
}
index.ts
// helper sleep method
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// ✅ Good: The `Promise.race` is surrounded by a `step.do`, ensuring deterministic caching behavior.
		const race_return = await step.do("Promise step", async () => {
			return await Promise.race([
				step.do("Promise first race", async () => {
					await sleep(1000);
					return "first";
				}),
				step.do("Promise second race", async () => {
					return "second";
				}),
			]);
		});

		await step.sleep("Sleep step", "2 hours");

		return await step.do("Another step", async () => {
			// This step will return `second` because the `Promise.race` was surround by the `step.do` method.
			return race_return;
		});
	}
}

ID экземпляров уникальны

Workflow ID экземпляров уникальны в пределах Workflow. ID является уникальным идентификатором, который связывает логи, метрики, состояние и статус запуска с конкретным экземпляром, даже после его завершения. Если бы повторное использование ID было разрешено, было бы трудно понять, к какому экземпляру относится ID экземпляра Workflow: выполнялся ли он вчера, на прошлой неделе или сегодня.

Также возникнет проблема, если вы захотите запустить несколько разных экземпляров Workflow с разными входные параметры для того же ID пользователя, поскольку вам сразу потребуется определить новое сопоставление ID.

Если вам нужно связать несколько экземпляров с определённым пользователем, продавцом или другим ID «клиента» в вашей системе, рассмотрите использование составного ID либо случайно сгенерированных ID с сохранением соответствия в базе данных, например в D1.

index.js
// This is in the same file as your Workflow definition
export default {
	async fetch(req, env) {
		// 🔴 Bad: Use an ID that isn't unique across future Workflow invocations
		let userId = getUserId(req); // Returns the userId
		let badInstance = await env.MY_WORKFLOW.create({
			id: userId,
			params: payload,
		});

		// ✅ Good: use an ID that is unique
		// e.g. a transaction ID, order ID, or task ID are good options
		let instanceId = getTransactionId(); // e.g. assuming transaction IDs are unique
		// or: compose a composite ID and store it in your database
		// so that you can track all instances associated with a specific user or merchant.
		instanceId = `${getUserId(req)}-${crypto.randomUUID().slice(0, 6)}`;
		let { result } = await addNewInstanceToDB(userId, instanceId);
		let goodInstance = await env.MY_WORKFLOW.create({
			id: instanceId,
			params: payload,
		});

		return Response.json({
			id: goodInstance.id,
			details: await goodInstance.status(),
		});
	},
};
index.ts
// This is in the same file as your Workflow definition
export default {
	async fetch(req: Request, env: Env): Promise<Response> {
		// 🔴 Bad: Use an ID that isn't unique across future Workflow invocations
		let userId = getUserId(req); // Returns the userId
		let badInstance = await env.MY_WORKFLOW.create({
			id: userId,
			params: payload,
		});

		// ✅ Good: use an ID that is unique
		// e.g. a transaction ID, order ID, or task ID are good options
		let instanceId = getTransactionId(); // e.g. assuming transaction IDs are unique
		// or: compose a composite ID and store it in your database
		// so that you can track all instances associated with a specific user or merchant.
		instanceId = `${getUserId(req)}-${crypto.randomUUID().slice(0, 6)}`;
		let { result } = await addNewInstanceToDB(userId, instanceId);
		let goodInstance = await env.MY_WORKFLOW.create({
			id: instanceId,
			params: payload,
		});

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

await ваши шаги

При вызове step.do или step.sleep, используйте await чтобы не допустить появления ошибок и состояний гонки в коде вашего Workflow.

Если вы не вызываете await step.do или await step.sleep, вы создаёте висящий Promise. Это происходит, когда Promise создаётся, но не awaitчто может привести к ошибкам и состояниям гонки.

Это происходит, если вы не используете await ключевое слово или не соедините в цепочку .then() методы для обработки результата Promise. Например, вызов fetch(GITHUB_URL) без ожидания ответа приведёт к тому, что последующий код выполнится немедленно, независимо от того, завершился ли fetch. Это может вызвать такие проблемы, как преждевременное логирование, поглощение исключений (из-за которого Workflow не завершается) и потерю возвращаемых значений (состояния).

index.js
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// 🔴 Bad: The step isn't await'ed, and any state or errors is swallowed before it returns.
		const badIssues = step.do(`fetch issues from GitHub`, async () => {
			// The step will return before this call is done
			let issues = await getIssues(event.payload.repoName);
			return issues;
		});

		// ✅ Good: The step is correctly await'ed.
		const goodIssues = await step.do(`fetch issues from GitHub`, async () => {
			let issues = await getIssues(event.payload.repoName);
			return issues;
		});

		// Rest of your Workflow goes here!
	}
}
index.ts
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// 🔴 Bad: The step isn't await'ed, and any state or errors is swallowed before it returns.
		const badIssues = step.do(`fetch issues from GitHub`, async () => {
			// The step will return before this call is done
			let issues = await getIssues(event.payload.repoName);
			return issues;
		});

		// ✅ Good: The step is correctly await'ed.
		const goodIssues = await step.do(`fetch issues from GitHub`, async () => {
			let issues = await getIssues(event.payload.repoName);
			return issues;
		});

		// Rest of your Workflow goes here!
	}
}

Аккуратно используйте условную логику

Вы можете использовать if операторов, циклов и другой управляющей логики вне шагов. Однако условия должны основываться на детерминированные значения : либо значения из event.payload или возвращаемых значений предыдущих шагов. Недетерминированные условия (такие как Math.random() или Date.now()) за пределами шагов может привести к непредсказуемому поведению при перезапуске Workflow.

index.js
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		const config = await step.do("fetch config", async () => {
			return await this.env.KV.get("feature-flags", { type: "json" });
		});

		// ✅ Good: Condition based on step output (deterministic)
		if (config.enableEmailNotifications) {
			await step.do("send email", async () => {
				// Send email logic
			});
		}

		// ✅ Good: Condition based on event payload (deterministic)
		if (event.payload.userType === "premium") {
			await step.do("premium processing", async () => {
				// Premium-only logic
			});
		}

		// 🔴 Bad: Condition based on non-deterministic value outside a step
		// This could behave differently if the Workflow restarts
		if (Math.random() > 0.5) {
			await step.do("maybe do something", async () => {});
		}

		// ✅ Good: Wrap non-deterministic values in a step
		const shouldProcess = await step.do("decide randomly", async () => {
			return Math.random() > 0.5;
		});
		if (shouldProcess) {
			await step.do("conditionally do something", async () => {});
		}
	}
}
index.ts
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		const config = await step.do("fetch config", async () => {
			return await this.env.KV.get("feature-flags", { type: "json" });
		});

		// ✅ Good: Condition based on step output (deterministic)
		if (config.enableEmailNotifications) {
			await step.do("send email", async () => {
				// Send email logic
			});
		}

		// ✅ Good: Condition based on event payload (deterministic)
		if (event.payload.userType === "premium") {
			await step.do("premium processing", async () => {
				// Premium-only logic
			});
		}

		// 🔴 Bad: Condition based on non-deterministic value outside a step
		// This could behave differently if the Workflow restarts
		if (Math.random() > 0.5) {
			await step.do("maybe do something", async () => {});
		}

		// ✅ Good: Wrap non-deterministic values in a step
		const shouldProcess = await step.do("decide randomly", async () => {
			return Math.random() > 0.5;
		});
		if (shouldProcess) {
			await step.do("conditionally do something", async () => {});
		}
	}
}

Пакетная обработка нескольких вызовов Workflow

При создании нескольких экземпляров Workflow используйте createBatch метод для группировки вызовов. Это позволяет создавать несколько инстансов Workflow в одном запросе, что уменьшает количество запросов к Workflows API. Однако каждый отдельный инстанс в группе всё равно будет учитываться в лимит частоты создания. В отличие от create, createBatch идемпотентен: если существующий инстанс с тем же ID всё ещё находится в своём лимит хранения, он будет пропущен и не войдёт в возвращаемый массив.

index.js
export default {
	async fetch(req, env) {
		let instances = [
			{ id: "user1", params: { name: "John" } },
			{ id: "user2", params: { name: "Jane" } },
			{ id: "user3", params: { name: "Alice" } },
			{ id: "user4", params: { name: "Bob" } },
		];

		// 🔴 Bad: Create them one by one, which is more likely to hit creation rate limits.
		for (let instance of instances) {
			await env.MY_WORKFLOW.create({
				id: instance.id,
				params: instance.params,
			});
		}

		// ✅ Good: Batch calls together
		// This improves throughput.
		let createdInstances = await env.MY_WORKFLOW.createBatch(instances);
		return Response.json({ instances: createdInstances });
	},
};
index.ts
export default {
	async fetch(req: Request, env: Env): Promise<Response> {
		let instances = [
			{ id: "user1", params: { name: "John" } },
			{ id: "user2", params: { name: "Jane" } },
			{ id: "user3", params: { name: "Alice" } },
			{ id: "user4", params: { name: "Bob" } },
		];

		// 🔴 Bad: Create them one by one, which is more likely to hit creation rate limits.
		for (let instance of instances) {
			await env.MY_WORKFLOW.create({
				id: instance.id,
				params: instance.params,
			});
		}

		// ✅ Good: Batch calls together
		// This improves throughput.
		let createdInstances = await env.MY_WORKFLOW.createBatch(instances);
		return Response.json({ instances: createdInstances });
	},
};

Ограничивайте тайм-ауты 30 минутами или меньше

При задании Тайм-аут WorkflowStep, убедитесь, что её продолжительность не превышает 30 минут. Если для вашего сценария нужен тайм-аут больше 30 минут, рассмотрите использование step.waitForEvent() взамен.

Размер возвращаемых значений непотоковых шагов не должен превышать 1 MiB

Непотоковый step.do() возвращаемое значение может сохраняться в объёме до 1 МиБ (2^20 байт). Если шаг возвращает структурированные данные, превышающие этот предел, шаг завершится со сбоем. Это распространённая проблема при получении больших ответов API или обработке больших файлов.

В JavaScript Workflows, ReadableStream<Uint8Array> это поддерживаемый сериализуемый тип возвращаемого значения для больших бинарных данных. При сохранении данных такого рода следует:

Обратите внимание: потоковые выходные данные всё равно учитываются в лимите хранилища экземпляра Workflow.

Если эти ограничения хранения всё равно вам не подходят, рассмотрите возможность хранения результатов шагов во внешнем хранилище (например, в R2) и сохранения ссылки на него.

index.js
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// 🔴 Bad: Returning a large response that may exceed 1 MiB
		const largeData = await step.do("fetch large dataset", async () => {
			const response = await fetch("https://api.example.com/large-dataset");
			return await response.json(); // Could exceed 1 MiB
		});

		// ✅ Good: Store large structured data externally and return a reference
		const dataRef = await step.do("fetch and store large dataset", async () => {
			const response = await fetch("https://api.example.com/large-dataset");
			const data = await response.json();
			// Store in R2 and return a reference
			await this.env.MY_BUCKET.put("dataset-123", JSON.stringify(data));
			return { key: "dataset-123" };
		});

		// Retrieve the data in a later step when needed
		const data = await step.do("process dataset", async () => {
			const stored = await this.env.MY_BUCKET.get(dataRef.key);
			return processData(await stored.json());
		});
	}
}
index.ts
export class MyWorkflow extends WorkflowEntrypoint {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// 🔴 Bad: Returning a large response that may exceed 1 MiB
		const largeData = await step.do("fetch large dataset", async () => {
			const response = await fetch("https://api.example.com/large-dataset");
			return await response.json(); // Could exceed 1 MiB
		});

		// ✅ Good: Store large structured data externally and return a reference
		const dataRef = await step.do("fetch and store large dataset", async () => {
			const response = await fetch("https://api.example.com/large-dataset");
			const data = await response.json();
			// Store in R2 and return a reference
			await this.env.MY_BUCKET.put("dataset-123", JSON.stringify(data));
			return { key: "dataset-123" };
		});

		// Retrieve the data in a later step when needed
		const data = await step.do("process dataset", async () => {
			const stored = await this.env.MY_BUCKET.get(dataRef.key);
			return processData(await stored.json());
		});
	}
}