Laravelのジョブをコミット後に実行する — afterCommitとOutboxの使い分け
この記事で扱う問題
注文を確定したら、出荷の依頼をキューのジョブに回し、確定の通知をイベントで送る。よくある処理です。素直に書くと、トランザクションの中でディスパッチします。
return $this->transaction->run(function () use ($id, $operatorId) {
$order = $this->orders->findOrFail($id);
$order->confirm($operatorId, $this->clock->now());
$this->orders->save($order);
RequestShipmentJob::dispatch($id->value); // Redis や SQS なら、この時点でワーカーが拾える
OrderConfirmed::dispatch($id->value); // 同期のリスナーは、この場で動く
return ConfirmOrderResult::from($order);
});
例えば、次のようなことが起こり得ます。
- ジョブはディスパッチした瞬間にキューに積まれます。ワーカーが空いていれば、コミットより先に動きます。ジョブが注文を読み直すなら、まだ確定前の状態を読みます。上の例では ID の文字列を渡していますが、モデルを
SerializesModelsで受け取るジョブなら、新しく作った行は見つからず、ModelNotFoundExceptionで失敗します（$deleteWhenMissingModelsを付けていれば、何も残さず消えます） - ディスパッチの後で例外が出てロールバックされても、キューに積んだジョブは戻りません。確定していない注文の出荷依頼が、配送業者に届きます
- 同期のリスナーは、トランザクションの中で動きます。リスナーがメールを送れば、ロールバックされても送られたままです。メールサーバーが遅ければ、その間ずっと注文の行のロックを持ち続けます
開発環境では、キューが sync だったり、トランザクションがすぐ終わったりして、ほとんど表に出ません。負荷が上がってワーカーが空き、トランザクションが長くなったときに、たまに起きる不具合として現れます。
逆に、ディスパッチをトランザクションの外に出せば順序は正しくなりますが、今度は「コミットしたのに、ディスパッチする前にプロセスが止まった」ときに、依頼が消えます。
判断
ジョブとイベントは、コミットされたときだけ、コミットの後に動くようにそろえます。そのうえで、消えると困る依頼だけは、業務の更新と同じトランザクションで DB に記録します。
この記事のコードは Laravel 10 以降を前提にしています（ShouldDispatchAfterCommit と ShouldQueueAfterCommit は 10.x の途中で追加されました）。
キューの接続で after_commit を有効にする
config/queue.php の接続ごとに、after_commit を true にします（Laravel の既定の設定ファイルでは false です）。
'redis' => [
'driver' => 'redis',
'connection' => env('REDIS_QUEUE_CONNECTION', 'default'),
'queue' => env('REDIS_QUEUE', 'default'),
'retry_after' => 90,
'block_for' => null,
'after_commit' => true,
],
これで、トランザクションの中でディスパッチしたジョブは、すぐには積まれず、トランザクションがコミットされた後に積まれます。ロールバックされれば捨てられます。トランザクションが入れ子になっていれば、一番外側のコミットまで待ちます。トランザクションの外でディスパッチしたジョブは、これまでどおりすぐに積まれます。
接続の設定とは別に、ジョブごとに ->afterCommit() を付けることもできます。また、ジョブやキューに入るリスナー・メール・通知のクラスに ShouldQueueAfterCommit を実装しておけば、接続の設定に関係なく、コミット後に積まれます。
この仕組みが追いかけるのは、Laravel の DB 接続で始めたトランザクションだけです。TransactionManager の実装は、DB::transaction() か、接続の transaction() を使う必要があります。PDO を直接使って始めたトランザクションは、Laravel からは見えません。
イベントは、イベントの側でコミット後にする
イベントには ShouldDispatchAfterCommit を実装します。イベント自体がコミット後に発行されるので、同期のリスナーも、キューに入るリスナーも、コミットの後に動きます。
use Illuminate\Contracts\Events\ShouldDispatchAfterCommit;
use Illuminate\Foundation\Events\Dispatchable;
final class OrderConfirmed implements ShouldDispatchAfterCommit
{
use Dispatchable;
public function __construct(public readonly string $orderId)
{
}
}
Eloquent のモデルイベントや、パッケージが発行するイベントのように、イベントのクラスを変えられない場合は、受け取る側で指定します。キューに入るリスナーには ShouldQueueAfterCommit、オブザーバーには ShouldHandleEventsAfterCommit を実装します。
UseCase からは、依頼する手段を Interface で受け取る
私は、UseCase から Laravel のジョブやイベントを直接ディスパッチしません。業務ルールを Laravel に依存しない場所に置くためと、「いつ、どう依頼するか」を実装の側で選べるようにするためです。UseCase が知るのは、「出荷を依頼する」という操作だけです。
interface ShipmentRequestQueue
{
/**
* 業務の更新と同じトランザクションの中で呼ぶ。
* 依頼が実際に出るのは、トランザクションがコミットされたときだけ。
*/
public function enqueue(OrderId $orderId): void;
}
final class ConfirmOrder
{
public function __construct(
private OrderRepository $orders,
private TransactionManager $transaction,
private Clock $clock,
private ShipmentRequestQueue $shipmentRequests,
) {
}
public function execute(OrderId $id, int $operatorId): ConfirmOrderResult
{
return $this->transaction->run(function () use ($id, $operatorId) {
$order = $this->orders->findOrFail($id);
$order->confirm($operatorId, $this->clock->now());
$this->orders->save($order);
$this->shipmentRequests->enqueue($id);
return ConfirmOrderResult::from($order);
});
}
}
「コミットされたときだけ依頼が出る」ことは、Interface の約束として書いておきます。実装は2つ考えられます。
1つ目は、ジョブを afterCommit() 付きでディスパッチする実装です。接続の設定が変わっても挙動が変わらないように、ここでは明示します。
use Illuminate\Contracts\Bus\Dispatcher;
final class AfterCommitShipmentRequestQueue implements ShipmentRequestQueue
{
public function __construct(private Dispatcher $bus)
{
}
public function enqueue(OrderId $orderId): void
{
$this->bus->dispatch((new RequestShipmentJob($orderId->value))->afterCommit());
}
}
消えると困る依頼は、同じトランザクションで記録する（Outbox）
2つ目は、依頼を DB のテーブルに記録し、別の処理がそこからジョブをディスパッチする実装です。いわゆる Outbox パターンです。
use Illuminate\Database\ConnectionInterface;
final class OutboxShipmentRequestQueue implements ShipmentRequestQueue
{
// TransactionManager と同じ接続を受け取る。別の接続だと、業務の更新と同じトランザクションに入らない
public function __construct(private ConnectionInterface $db)
{
}
public function enqueue(OrderId $orderId): void
{
// shipment_request_outbox には order_id の一意制約がある。同じ注文の依頼は1行だけ残る
$this->db->table('shipment_request_outbox')->insertOrIgnore([
'order_id' => $orderId->value,
'created_at' => now(),
]);
}
}
記録は注文の確定と同じトランザクションに入るので、確定と依頼の記録は、必ずそろってコミットされるか、そろってロールバックされます。記録した依頼は、スケジューラから毎分動くコマンドがディスパッチします。
use Illuminate\Console\Command;
use Illuminate\Contracts\Bus\Dispatcher;
use Illuminate\Database\ConnectionInterface;
final class RelayShipmentRequests extends Command
{
protected $signature = 'outbox:relay-shipment-requests';
public function handle(ConnectionInterface $db, Dispatcher $bus): int
{
$rows = $db->table('shipment_request_outbox')
->whereNull('dispatched_at')
->orderBy('id')
->limit(100)
->get();
foreach ($rows as $row) {
// トランザクションの外なので、after_commit の設定に関係なく、すぐに積まれる
$bus->dispatch(new RequestShipmentJob($row->order_id));
// ここで止まると、次の実行で同じ依頼をもう一度ディスパッチする
$db->table('shipment_request_outbox')
->where('id', $row->id)
->update(['dispatched_at' => now()]);
}
return self::SUCCESS;
}
}
ディスパッチしてから dispatched_at を書くまでの間にプロセスが止まると、同じ依頼が2回ディスパッチされます。Outbox が保証するのは「少なくとも1回」です。そのため、ここで投げるジョブは何回実行しても結果が1回と同じになるように作っておく必要があります。RequestShipmentJob は「この注文の出荷依頼」から決まる Idempotency Key を配送業者に渡しているので、2回届いても二重には出荷されません。同じ理由で、コマンドが重なって動いても正しさは崩れませんが、無駄なディスパッチを減らすために、スケジューラでは withoutOverlapping() を付けます。
なぜそう判断したか
UseCase の後でディスパッチする方法を選ばなかった理由
$this->transaction->run() が返った後にディスパッチしても、順序は正しくなります。ただ、2つの点で頼りにくいと考えています。
1つは、「後でディスパッチする」ことを、UseCase を書く人が毎回覚えている必要があることです。もう1つは、入れ子です。この UseCase が別のトランザクションの中から呼ばれると、run() が返った時点では、外側のトランザクションはまだコミットされていません。「トランザクションの外」のつもりで書いた行が、実は中にあります。after_commit は一番外側のコミットまで待つので、呼ばれ方に左右されません。
after_commit だけで済ませない理由
after_commit で解決するのは、「コミットより先に動く」と「ロールバックされたのに動く」の2つです。コミットとキューへの書き込みは、別々のシステムへの書き込みのままです。
- コミットした直後、キューに積む前にプロセスが止まれば、依頼は消えます。デプロイ時の再起動でも起こり得ます
- キュー（Redis や SQS）に積むのに失敗すると、その例外は
transaction()から返ります。DB はコミット済みです。同じトランザクションで、それより後に予約していたほかのジョブやイベントも動きません。Controller は 500 を返し、利用者が再送すると、今度は「確定済みの注文は確定できない」で失敗します。利用者から見ると、失敗したはずの操作が実は済んでいた、という状態です
DB と S3 をまたぐ更新と同じで、2つのシステムを同時に更新することはできません。選べるのは、どちらの不整合を許容するかです。通知メールなら、まれに届かなくても大きな問題にはならないので、after_commit で十分です。出荷の依頼やポイントの付与のように、消えると利用者が困り、しかも誰も気づかないものには、Outbox でキューの障害と業務の更新を切り離します。Outbox なら、キューが落ちていても注文の確定は成功し、依頼はキューが戻ってから出ます。
データベースのキューを使う方法
キューのドライバに database を使い、業務のテーブルと同じ DB 接続に置いている場合は、事情が違います。after_commit を false のままにすると、ジョブの行は業務の更新と同じトランザクションで INSERT されます。コミットされるまでワーカーからは見えず、ロールバックされれば一緒に消えます。これは Outbox と同じ性質です。逆に、この接続で after_commit を true にすると、積むのがコミットの後になり、この性質を自分で手放すことになります。
ただ、ジョブの量が増えると、キューの読み書きが業務の DB の負荷になり、キューだけを別の基盤に移すこともできなくなります。私は、キューは Redis や SQS に置き、消えると困る依頼だけを Outbox で記録する、という分け方を選んでいます。
どこから設計過剰になるか
- 消えても後から気づける処理には、Outbox は要りません。キャッシュの作り直しや集計の更新は、次の更新や定期的な作り直しで追いつきます。通知メールも、多くは after_commit で足ります
- Outbox を作るなら、まずは依頼の種類ごとのテーブルとコマンドで始めます。どの種類の依頼も載せられる汎用の Outbox と配送の仕組みは、依頼の種類が増えてから考えます。コマンドが毎分動くので、依頼は最大で1分ほど遅れます。それが困る処理なら、Outbox に記録したうえで after_commit でもすぐにディスパッチし、コマンドは一定時間たっても
dispatched_atが空の行だけを拾う、という組み合わせもできます（その場合、即時のディスパッチで投げたジョブがdispatched_atを書く必要があります） - 処理全体がトランザクションを使っていないなら、after_commit を付けても何も変わりません。問題になるのは、トランザクションの中でディスパッチしている箇所だけです
- 既存のアプリで接続ごと after_commit を有効にするときは、1つの大きなトランザクションの中で大量のジョブをディスパッチしているバッチがないかを先に確かめます。コミットまでジョブはメモリに溜まり、コミットの瞬間にまとめて積まれます。こうした処理は、トランザクションを小さく分けるほうが先です
まとめ
- トランザクションの中でディスパッチしたジョブやイベントは、コミットより先に動いたり、ロールバックされても動いたりする
- キューの接続の after_commit と、イベントの
ShouldDispatchAfterCommitで、コミットされたときだけ後から動くようにそろえる - コミットとキューへの書き込みは別の書き込みのまま。消えると困る依頼は Outbox で同じトランザクションに記録し、ジョブを冪等にする