Cloudflareスタックで非同期ジョブをいい感じに管理したい
- 投稿した日
- 2026/08/13
- 更新した日
- 2026/08/13
- 読了まで
- 13.72分で読み終われます (8,235文字)
# はじめに
どうも、わたしです。
ここ数年で、Cloudflare Workersを取り巻く状況はかなり変わっているようで、Workersそのものだけでなく、QueuesやDurable Objects、D1、Analytics Engine等、Cloudflareスタックの中で使えるサービスが結構増えました。加えて、比較的安価に使えることもあり、Workersを本格的なバックエンドとして採用するケースも(わたしの周りでは)かなり増えたように思います。
利用者が増えれば当然知見も増えるもので、Durable Objectsをどう使うと嬉しいのか、D1に何を任せるべきなのか、Containersで何が解決可能なのか…といった話を見かけることが増えました。
そんな中で、Workersスイートで構築されたアプリケーションを運用していると、バックグラウンドジョブ周りでいくつか困ることが出てきました。
そのあたりをどうにかしようとして作り始めたのが、今回紹介するTsumugiです。
Cloudflare Workers / Durable Objects / Queues / D1を中心に、バックグラウンドジョブの投入、遅延実行、再試行、優先度、同時実行数制御、定期実行、ジョブ同士の依存関係などをまとめてブラックボックスとしていい感じに扱うためのジョブ管理システムです。最初から汎用的なジョブシステムを作ろうと思って始めたわけではないのですが、気がついたらそこそこ大きくなっていました。
そしてせっかく公開したのに思っていた以上に反応をもらえず、たいへん寂しいのでこの記事を書いています。
これは宣伝です。笑ってください。
# きっかけ
Tsumugiを作る直接のきっかけは、以前作ったMisskey向け匿名質問箱サービスのMewkです。
Mewkでは、バックグラウンドジョブの管理にKiribiというCloudflare Queues向けのジョブ管理システムを採用していました。Queuesのインターフェースを隠して単純な同期処理らしく扱えるようにしてくれるライブラリで、状態管理や再試行まで含めて面倒を見てくれるため、かなり便利に使っていました。
一方で、実際のサービスに載せて運用していると、少しずつKiribiの外側に欲しいものが増えていきました。特定の種類のジョブだけ同時実行数を厳密に制限したかったり、待機中のジョブに優先順位を付けたかったり、同じ対象に対するジョブを重複して積みたくなかったり、複数の処理に依存関係を持たせたかったり、みたいなやつです。
どれも個別には実装できますし、Kiribi自体も使いづらいライブラリだったわけではありません。が、必要な機能をアプリケーション側で補っていくと、Kiribiを使いながら別のジョブスケジューラを作っているような微妙な状態になってきました。
KiribiへContributeする方向も当然考えたものの、2年以上ほどメンテナンスが続いていませんでしたし、必要としていた変更も単純な機能追加というより、スケジューリングや状態管理の責務をどこに置くかという設計そのものに関わるものであったことも踏まえ、既存の設計へ無理に差し込むよりも現在のCloudflareスタックを前提に一度作り直した方がよいのではないか、と考えました。
有難いことに、Kiribiが登場した当時からCloudflare側の状況もかなり変わっているようで、Cloudflareスタックの中だけでより複雑な状態管理を成立させるための選択肢も増えており、その結果として作り始めたのがTsumugiです。
# Queuesだけでは少し足りない
Cloudflare Queuesは、consumerへの配送, 失敗時の再試行まで面倒を見てくれる便利なものです。単純な非同期処理であれば、(希に揮発することもありますが)Queueを直接使うだけで十分なことも多いです。
しかし、実際に使っていると、Queuesはあくまで配送のためのものであって、次にどのジョブを実行するべきかを判断するスケジューラとして使えない部分がつらくなってきます。
既にn件のジョブが実行されている間はn+1件目をまだ動かしたくないだとか、待機中のジョブの中でもpriorityの高いものを先に実行したいだとか、あるジョブは別のジョブが成功するまで開始させたくないみたいな場合ですね。
こういう細かな制御をしたい場合、今どういう状態なのかを見ながら次に何を実行するかを判別する実装を書いてあげる必要があります。
# Durable Objectで判別, Queuesで実行
Tsumugiでは、bindingごとにDurable Objectを置き、それをスケジューラ兼コーディネータとして利用しています。
Durable Objectは、待機中のジョブ、現在実行中のジョブ、実行可能時刻、priority、concurrency、rate limitなどを見ながら、次に何をdispatchするかを決定しています。
この部分をDurable Objectに委ねた主な理由として、整合性が欲しかったためです。たとえば「このジョブは最大3件まで同時実行する」と決めたのであれば、4件目は本当に実行されてほしくありません。D1上のカウンタを何となく増減させながら判定するよりも、ひとつのDurable Objectが状態を持ち、そこで判断を直列化した方がはるかに分かりやすいはずです。
ただし、Durable Object自身にジョブ本体を実行させているわけではありません。
たとえば10分かかる処理をDurable Objectから直接awaitした場合、その10分間ずっとDurable Object側のリクエストを生かしておくことになり、それはあまり嬉しくありません。
なのでTsumugiでは、Durable Objectは判断するだけで、実際の実行はQueuesに任せる、という形にしています。
# 理解せず単純に扱える
内部の話をすると、Durable Object, SQLite, Queues, D1, outbox, projectionなど色々なものが出てきます。
とはいえ、利用者側までその事情を意識する必要がないようにしたかったので、表側はなるべく普通のジョブライブラリっぽくしています。
たとえばメールを送るジョブならこんな感じで書けます。
import { Performer } from 'tsumugi/performer';
export class SendMail extends Performer<{ to: string }, void, {}, Env> {
async perform(payload: { to: string }): Promise<void> {
await this.env.MAILER.send(payload.to);
}
}
これをWorkerのトップレベルからexport
export * from './performers/index.js';
あとは
await tsumugi.enqueue(env, {
binding: 'SendMail',
payload: {
to: 'a@example.com',
},
});
とすればenqueueできます。
ちなみにbindingにSendMailを指定すると、payloadもPerformer側から型推論されます。
同じ情報を複数箇所に書くのは冗長ですごく嫌いなので、TsumugiではWorkerからexportされているPerformerをruntime上のregistryとして使いつつ、型も同じ場所から取るようにしています。
# ジョブらしいことは一通りできる(はず)
基本のenqueueに加えて、遅延実行も指定できます
await tsumugi.enqueue(env, {
binding: 'SendMail',
payload,
delayMs: 60_000,
});
priorityを付けることもできます
await tsumugi.enqueue(env, {
binding: 'SendMail',
payload,
priority: 10,
});
同じ対象に対するジョブを重複して作りたくない場合は、uniqueKeyも明示可能です
await tsumugi.enqueue(env, {
binding: 'SendMail',
payload,
uniqueKey: 'a@example.com',
});
このほかにも、bindingごとの同時実行数制御, rate limit, retry, 定期実行, remote performer, retry/cancellation, 実行履歴の追跡などがあります。また、複数のジョブに依存関係を持たせるFlowという仕組みもありますが、これは少し毛色が違うので後述します。が、全部ここで説明し始めるとそのままREADMEになってしまうので、細かい使い方はドキュメントを見てください。
ちゃんと書きました。
# ジョブ同士の依存関係も扱える
Tsumugiには、複数のジョブを依存関係ごとまとめて扱うFlowという仕組みがあります。
これは、ジョブを頂点、ジョブ同士の依存関係を辺とした有向非巡回グラフ(DAG)として処理全体を表現するものです。
たとえば画像を受け取って、まず解析を行い、その結果を使ってサムネイルの生成とメタデータの保存を並列に実行し、両方が終わったあとに公開処理を行う場合は、概念的には次のようなグラフになります。
この場合、公開処理をするは、サムネイルを生成するとメタデータを保存するの両方が完了するまで実行されません。
こうした処理自体は、Aの処理が成功したらBをenqueueし、Bが終わったらCとDをenqueueし、それぞれの完了状態をどこかに記録して、両方が終わったらEをenqueueすることでQueues単体でも一応実現可能ではあります。しかし、どのジョブが完了したのか、次に実行可能になるジョブはどれなのか、途中のジョブが失敗した場合に後続をどう扱うのか、といったことまで自分で管理しなければなりません。
Tsumugiでは、この依存関係そのものをジョブシステム側の状態として管理します。あるジョブが完了すると、そのジョブを依存先としている後続ジョブについて、すべての依存関係が解決したかをTsumugi側で判定し、実行可能になったものだけを順次スケジューリングします。
Temporalのような汎用的で大規模なworkflow engineを目指しているわけではありませんが、バックグラウンド処理を書いていると「AとBが終わったらCを動かしたい」くらいの依存関係はわりと普通に出てきます。そうした関係を個々のPerformerの中へ書き散らすのではなく、ジョブ同士の関係そのものをTsumugiに管理させられる、というのがFlowです。
# Cloudflareスタック
TsumugiはCloudflare専用です。
少なくとも今のところ、AWSでも動くとかGCPでも動くとか、そういう方向はあまり考えていません。Durable ObjectsもQueuesもD1も普通に使います。
これは意図的なもので、どこでも動くライブラリを作ろうとすると、どうしても各環境の一番弱い共通部分へ寄せた抽象化になってしまいます。折角CloudflareにはDurable Objectsのような特殊で便利なプリミティブがあるのに、それを使わず一般化するのはもったいないと思っています。
逆に言うとTsumugiは、Cloudflareでしか動かなくていいので、Cloudflareではちゃんと気持ちよく動くものを目指しています。
こういうことばかり言うと気持ち悪いのでデメリットっぽいものも紹介しておくと、Workers Paidが必要ですし、D1やQueuesの準備も必要です。Kiribiと比べて依存するサービス数も増えるのでコスト的には劣っています。単純に「重要度の低いWebhookを受け取って後で1回だけ処理したい」程度であれば、Tsumugiを入れる必要はありませんし、なんならQueuesを直接使った方がたぶんしあわせになれます。
# Kiribiとの違い
TsumugiはKiribiからかなり影響を受けていますが、結果として設計はだいぶ違うものになりました。
Kiribiは、(わたしの解釈に誤りがなければ)Cloudflare Queuesにジョブ管理の仕組みを足すという方向のライブラリです。それに対してTsumugiでは、Queuesをジョブシステムそのものではなく、実行時の配送手段として扱っています。
Kiribiの方が構成は単純ですし、Queues自身が持っているスケーラビリティの恩恵を素直に受けることができます。一方でTsumugiは、スケジューリングの判断を自分で握っている分、priorityや正確なconcurrency制御、Flow、rate limitなどをひとつの状態モデルの上で扱いやすくなっています。
ですので、TsumugiがKiribiの完全な上位互換だとは思っていません。そもそも、単純なジョブ管理ならKiribiの方が軽量ですし、Queuesをなるべくそのまま使いたいならKiribiの設計の方が自然に見えます。Tsumugiは、その先でもう少し制御したくなったときの選択肢として使っていただけると嬉しいです。
# 分散システムは嫌
ここまで色々と書いていますが、Tsumugiの内部は割とちゃんと分散システムです。もうなんか嫌ですね。
Tsumugiの内部では、Durable Object, Queues, Performer, D1と複数のコンポーネントを跨いでジョブを処理しています。そのため、正常系だけでなく、それぞれの境界で処理が失敗した場合に状態をどう収束させるかはかなり意識して設計しました。
たとえば、Durable Object側ではジョブをdispatchする状態まで進んだもののQueueへの送信に失敗した場合や、Performerの処理自体は成功したものの、その完了をDurable Objectへ通知する段階で失敗した場合などがあります。D1へのprojectionについても同様で、途中で失敗したからといってジョブそのものの状態まで不整合になってしまうのは避けるべきです。
Tsumugiでは、こうした問題を扱いやすくするために、ジョブのauthoritativeな状態と実際の配送、実行、read modelへの反映をそれぞれ分けています。
例として、retryについてはQueues側へ判断を任せるのではなく、Durable Object側をauthorityとして扱っています。また、D1はあくまでread modelなので、projectionが一時的に遅れたり失敗したりしても、それによってジョブの実行状態そのものが変わることはないようにしています。配送保証についてもジョブ単位で扱えるようにしていて、at-most-onceを必要とするジョブについては、実行前のclaimを含めて通常のジョブとは異なる扱いをしています。
このあたりは機能だけを見るとあまり目立たない部分ですが、ジョブ管理システムとして実際に使ううえではかなり重要だと考えていて、正常系だけでなく、Queueへの配送やWorkerの実行、状態更新のどこかが途中で失敗した場合にも、最終的に説明可能な状態へ戻せることを意識しました。
# 導入周りもまとめて
Tsumugiはnpm packageとして公開しているほか、初期セットアップ用のCLI, D1のmigration, 管理画面, REST API, ドキュメント, 実装例等もまとめて提供しています。
導入自体は、packageを追加したあとにtsumugi initを実行するところから始められます。
pnpm add tsumugi
npx tsumugi init
tsumugi initでは、Tsumugiが利用するD1 DatabaseやQueueの作成、wrangler設定やソースコードの雛形生成、migrationの適用などを行います。
Tsumugiは内部でDurable Objects、Queues、D1と複数のCloudflareサービスを利用するため、これらをすべて手作業で設定する形にすると導入時に確認する項目がかなり増えてしまいます。そのため、最初に必要になる構成についてはできるだけCLI側で生成するようにしました。
加えて、ジョブの状態を確認するためのダッシュボードもpackageに含まれており、別途管理画面を用意しなくとも、ジョブの一覧や実行履歴, エラーの確認, retry/cancelなどの操作が可能です。REST APIも提供されるため、必要であれば独自の管理ツールなどから操作することもできるかと思います。
内部構成はそれなりに複雑ですが、Tsumugiを利用するにあたっては、その中身を意識せず、ブラックボックスとして容易に扱えることを意識しています。
# おわりに
そんなわけで、Cloudflareスタック向けのジョブ管理システム、Tsumugiを作りました。
もともとはKiribiを使っていて、実運用の中で「もう少しここを制御したい」と思ったことが始まりでした。Kiribi自体は便利でしたが、必要な機能が増えるにつれて既存の設計の外側へはみ出す部分が増え、さらに当時すでに2年以上メンテナンスが続いていなかったこともあり、今のCloudflareスタックを前提に一度作り直してみることにしました。
Kiribiが登場した頃から時間が経ち、Cloudflareのサービスも増え、Workersだけでシステムを組むこと自体もずいぶん現実的になりました。であれば、今あるプリミティブを前提に、もう一度ジョブ管理システムを考えてみよう。というのがTsumugiです。
Cloudflare Queuesを使っていて、「配送はQueuesでいいけれど、ジョブの実行順や状態はもう少し自分で管理したい」だとか、「同時実行数やpriority、依存関係まで含めて扱いたい」みたいなものにあたりに心当たりがあれば、試してもらえると嬉しいです。
まだ利用者もほとんどいないので、設計上変なところ、不具合、使いづらいAPI、こういうものが欲しい、などあれば雑にでもIssueを投げてもらえると助かります。
ここまで読んで、まあちょっと面白いことをやっているじゃん くらいに思っていただけたのであれば、Starをひとつ置いていってもらえるとわたしが喜びます。
それでは〜