Workflow には 1 つ以上のステップがあります。各ステップは自己完結し、個別に再試行できる Workflow の構成要素です。ステップは(任意で)状態を出力でき、ネットワークやインフラの問題で Workflow が失敗しても、そのステップから永続化して再開できます。
耐障害性が高く、正しい Workflows を作るための短いガイドです。
ステップは複数回再試行されることがあるので、ステップは(理想的には)べき等にしてください。べき等性とは、操作(ここではステップ)を何度適用しても、最初の適用以降の結果が変わらない論理的な性質です。
たとえば、顧客に課金する Workflow があり、誤って二重課金したくないとします。課金する前に、すでに課金済みかどうかを確認してください。
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 }),
},
);
},
);
}
}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 / バインディング呼び出しは最小限にします(べき等性を示すために複数回呼ぶ必要がある場合を除く)。
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}`);
});
}
}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 負荷の高い処理をやりすぎないでください。エンジンが再起動することがあり、そのステップの先頭からやり直します。
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}`);
});
}
}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、再試行、イベント)ハイバーネートできるときに発生します。
そのため、ステップの外に状態を置かないでください。
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}`);
},
);
}
}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 の戻り値だけで構成してください。
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}`);
},
);
}
}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}`);
},
);
}
}ステップの外に副作用のあるコードを書くことは推奨しません。繰り返してよい処理でない限り避けてください。インスタンス実行中に Workflow エンジンが再起動することがあるためです。エンジンが再起動してもステップ内のロジックは保たれますが、ステップ外のロジックは重複する可能性があります。
たとえば、Workflow のステップ外の console.log() は、エンジン再起動時にログが 2 回出力されることがあります。
一方、データベース接続のようなシリアライズできないリソースを扱うロジックは、ステップの外で実行してください。step.do の外の操作は、Workflows のインスタンスライフサイクルの性質上、複数回繰り返されることがあります。
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;
},
);
}
}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 の再起動をまたいで永続化されません。
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
},
);
}
}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 内の「キャッシュキー」として働きます。
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);
});
}
}
}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);
});
}
}
}Workflows では、ステップを同時実行する方法として、Promise.race() や Promise.any() の中でステップを使えます。ただし、いくつか注意点があります。
Workflows のインスタンスライフサイクルの性質上、Promise 内のステップは完了するまで走ります。最初の通過で返されたステップが、実際にキャッシュされたステップではないことがあります。ステップは名前でキャッシュされる ためです。
// 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;
});
}
}// 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() で囲むことを推奨します。そうすると、複数回の通過でもキャッシュが一貫します。
// 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;
});
}
}// 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;
});
}
}Workflow の インスタンス ID は、Workflow ごとに一意です。ID は、完了後も含めて、ログ、メトリクス、状態、ステータスを特定のインスタンスに紐づける一意の識別子です。ID の再利用を許すと、その Workflow インスタンス ID が昨日、先週、今日のどれを指すのか分かりにくくなります。
同じユーザー ID に対して、異なる 入力パラメーター で複数の Workflow インスタンスを動かしたい場合にも問題になります。すぐに新しい ID の対応を決める必要があるためです。
特定のユーザー、加盟店、その他の「顧客」ID に複数インスタンスを紐づけたい場合は、複合 ID を使うか、ランダム生成した ID を D1 などのデータベースに保存することを検討してください。
// 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(),
});
},
};// 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(),
});
},
};step.do や step.sleep を呼ぶときは、await を使ってください。使わないと、Workflow コードにバグや競合状態が入りやすくなります。
await step.do や await step.sleep を呼ばないと、ダングリング Promise になります。Promise が作られても適切に await されない状態で、バグや競合状態の原因になります。
await キーワードを使わない、または結果を扱う .then() をつなげないときに起きます。たとえば、fetch(GITHUB_URL) のレスポンスを待たないと、fetch の完了を待たずに後続コードがすぐ実行されます。その結果、早すぎるログ出力、例外が飲み込まれて Workflow が終了しない、戻り値(状態)の紛失などが起きることがあります。
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!
}
}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 が再起動したときに想定外の動きをすることがあります。
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 () => {});
}
}
}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 インスタンスを作るときは、createBatch メソッドで起動をまとめてください。1 回のリクエストで複数インスタンスを作れるので、Workflows API へのリクエスト数を減らせます。ただし、バッチ内の各インスタンスは、引き続き 作成レート制限 にカウントされます。create と異なり、createBatch はべき等です。同じ ID の既存インスタンスがまだ 保持期間 内ならスキップされ、戻り値の配列から除外されます。
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 });
},
};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 });
},
};WorkflowStep のタイムアウト を設定するときは、長さを 30 分以下にしてください。30 分を超えるタイムアウトが必要な場合は、代わりに step.waitForEvent() の利用を検討してください。
非ストリームの step.do() 戻り値は、最大 1 MiB(2^20 バイト)まで永続化できます。構造化データがこの上限を超えると、ステップは失敗します。大きな API レスポンスを取得したり、大きなファイルを処理したりするときに起きやすい問題です。
JavaScript の Workflows では、より大きなバイナリ出力向けに、シリアライズ可能な戻り値の型として ReadableStream<Uint8Array> をサポートしています。この種の出力を永続化するときは、次のようにしてください。
-
ステップのコールバックから新しいストリームを返す。
-
個々のチャンクは 16 MB 未満にする。
-
ロック済みのストリームや、すでに読み取ったストリームを返さない。
-
ステップから返されたストリームだけを使う。
BYOB ストリームと BYOB リーダーはサポートされません。 :::
ストリーム出力も、Workflow インスタンスのストレージ上限に含まれます。
これらのストレージ上限でも足りない場合は、ステップ出力を外部(たとえば R2)に保存し、その参照だけを残すことを検討してください。
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());
});
}
}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 を組み合わせるときに役立ちます。