Skip to content

非公式本サイトは非公式の日本語ドキュメントであり、Cloudflare 公式サイトではありません。最新情報はdevelopers.cloudflare.comをご確認ください。

Workflows のルール

最終更新 Markdown で表示Agent セットアップ

Workflow には 1 つ以上のステップがあります。各ステップは自己完結し、個別に再試行できる Workflow の構成要素です。ステップは(任意で)状態を出力でき、ネットワークやインフラの問題で Workflow が失敗しても、そのステップから永続化して再開できます。

耐障害性が高く、正しい Workflows を作るための短いガイドです。

API / バインディングの呼び出しをべき等にする

ステップは複数回再試行されることがあるので、ステップは(理想的には)べき等にしてください。べき等性とは、操作(ここではステップ)を何度適用しても、最初の適用以降の結果が変わらない論理的な性質です。

たとえば、顧客に課金する Workflow があり、誤って二重課金したくないとします。課金する前に、すでに課金済みかどうかを確認してください。

index.jsjs
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.tsts
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 の障害やネットワークエラーなどがあっても、ロジックの耐久性が高まります。

トランザクション、または作業の単位と考えることもできます。

  • ✅ ステップあたりの API / バインディング呼び出しは最小限にします(べき等性を示すために複数回呼ぶ必要がある場合を除く)。
index.jsjs
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.tsts
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 全体の耐久性が想定より低くなり、未定義の動作に出会うことがあります。次のルールに従えば避けられます。

  • 🔴 ロジック全体を 1 つのステップにまとめないでください。
  • 🔴 別々のサービスを同じステップから呼び出さないでください(べき等性を示すために必要な場合を除く)。
  • 🔴 同じステップでサービス呼び出しを増やしすぎないでください(べき等性を示すために必要な場合を除く)。
  • 🔴 1 つのステップ内で CPU 負荷の高い処理をやりすぎないでください。エンジンが再起動することがあり、そのステップの先頭からやり直します。
index.jsjs
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.tsts
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 はハイバーネートし、メモリ上の状態をすべて失うことがあります。エンジンが保留中の作業がないと判断し、起きる必要があるまで(sleep、再試行、イベント)ハイバーネートできるときに発生します。

そのため、ステップの外に状態を置かないでください。

index.jsjs
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.tsts
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.jsjs
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.tsts
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 エンジンが再起動することがあるためです。エンジンが再起動してもステップ内のロジックは保たれますが、ステップ外のロジックは重複する可能性があります。

たとえば、Workflow のステップ外の console.log() は、エンジン再起動時にログが 2 回出力されることがあります。

一方、データベース接続のようなシリアライズできないリソースを扱うロジックは、ステップの外で実行してください。step.do の外の操作は、Workflows のインスタンスライフサイクルの性質上、複数回繰り返されることがあります。

index.jsjs
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.tsts
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;
			},
		);
	}
}

受け取ったイベントを変更しない

Workflow の run メソッドに渡される event はイミュータブルです。イベントに加えた変更は、ステップ間や Workflow の再起動をまたいで永続化されません。

index.jsjs
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.tsts
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.jsjs
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.tsts
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.jsjs
// 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 non-deterministic 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.tsts
// 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 non-deterministic 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.jsjs
// 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.tsts
// 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 の再利用を許すと、その Workflow インスタンス ID が昨日、先週、今日のどれを指すのか分かりにくくなります。

同じユーザー ID に対して、異なる 入力パラメーター で複数の Workflow インスタンスを動かしたい場合にも問題になります。すぐに新しい ID の対応を決める必要があるためです。

特定のユーザー、加盟店、その他の「顧客」ID に複数インスタンスを紐づけたい場合は、複合 ID を使うか、ランダム生成した ID を D1 などのデータベースに保存することを検討してください。

index.jsjs
// 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.tsts
// 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.dostep.sleep を呼ぶときは、await を使ってください。使わないと、Workflow コードにバグや競合状態が入りやすくなります。

await step.doawait step.sleep を呼ばないと、ダングリング Promise になります。Promise が作られても適切に await されない状態で、バグや競合状態の原因になります。

await キーワードを使わない、または結果を扱う .then() をつなげないときに起きます。たとえば、fetch(GITHUB_URL) のレスポンスを待たないと、fetch の完了を待たずに後続コードがすぐ実行されます。その結果、早すぎるログ出力、例外が飲み込まれて Workflow が終了しない、戻り値(状態)の紛失などが起きることがあります。

index.jsjs
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.tsts
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.jsjs
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.tsts
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 メソッドで起動をまとめてください。1 回のリクエストで複数インスタンスを作れるので、Workflows API へのリクエスト数を減らせます。ただし、バッチ内の各インスタンスは、引き続き 作成レート制限 にカウントされます。create と異なり、createBatch はべき等です。同じ ID の既存インスタンスがまだ 保持期間 内ならスキップされ、戻り値の配列から除外されます。

index.jsjs
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.tsts
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 MiB(2^20 バイト)まで永続化できます。構造化データがこの上限を超えると、ステップは失敗します。大きな API レスポンスを取得したり、大きなファイルを処理したりするときに起きやすい問題です。

JavaScript の Workflows では、より大きなバイナリ出力向けに、シリアライズ可能な戻り値の型として ReadableStream<Uint8Array> をサポートしています。この種の出力を永続化するときは、次のようにしてください。

  • ステップのコールバックから新しいストリームを返す。

  • 個々のチャンクは 16 MB 未満にする。

  • ロック済みのストリームや、すでに読み取ったストリームを返さない。

  • ステップから返されたストリームだけを使う。

BYOB ストリームと BYOB リーダーはサポートされません。 :::

ストリーム出力も、Workflow インスタンスのストレージ上限に含まれます。

これらのストレージ上限でも足りない場合は、ステップ出力を外部(たとえば R2)に保存し、その参照だけを残すことを検討してください。

index.jsjs
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.tsts
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());
		});
	}
}

関連リソース

  • Workers のベストプラクティス: リクエスト処理、オブザーバビリティ、セキュリティのコードパターンです。Workflows を起動する Workers にも当てはまります。
  • Durable Objects のルール: 状態を持ち、協調するアプリケーション向けのベストプラクティスです。Durable Objects と Workflows を組み合わせるときに役立ちます。

役に立ちましたか?