メッセージの配信と処理の再開
イベント全体の状態だけでなく、受信先と工程ごとに完了を記録する理由を説明します。
説明
一つのイベントを複数の受信先へ配る場合、それぞれの完了状態は独立しています。また、決済と発送のような複数工程をまとめて再実行すると、完了済みの工程まで繰り返すおそれがあります。
SQLiteにbillingとmailへの配信状態を保存し、一方だけ失敗させます。その後、決済済みで発送に失敗した処理を再開し、決済記録が1件のまま発送を完了できることを確認します。
イベント全体の状態だけでなく、受信先と工程ごとに完了を記録する理由を説明します。
1二つの受信先へ配信する
order-7の配信行をbillingとmailへ1件ずつ作ります。主キーは受信先とイベントIDの組です。
from workflow import Delivery工程と配信状態の試料を読み込む
flow = Delivery()保存先を用意する
flow.publish("order-7")二つの受信先へ同じイベントを登録する両方pendingです。一つの受信先が取れば他から消える仕事キューとは異なり、各受信先に処理の記録があります。
2一方だけ完了する
billingの処理を完了し、mailは失敗させます。
flow.consume("billing", "order-7")billingの処理を確定する
flow.consume("mail", "order-7", fail=True)mailだけ失敗させるbillingはdone、mailはpendingのままです。全受信先が成功した状態ではありません。
3イベントが再度届く
同じイベントをもう一度publishします。既存の行を初期状態へ戻さず、不足する行だけを登録します。
flow.publish("order-7")重複した配信を受け付けるbillingのdoneが保持され、mailだけが未完了です。再配送によって完了済みの情報を消してはいけません。
4失敗した受信先を再処理する
mailの処理を成功させます。
flow.consume("mail", "order-7")未完了のmailだけを確定する二つともdoneになりました。billingを再実行しなくても、mailの失敗を解消できます。
5決済後に発送が失敗する
次は工程を持つ処理です。chargeの保存と工程完了の記録を同じトランザクションで確定した後、shipを失敗させます。
flow.run("job-1", ship_ok=False)決済を確定し、発送で失敗させる
flow.status()工程と金額の保存結果を読むchargeだけが完了し、金額100の記録が1件あります。処理全体はまだ完了していません。
6完了済みの工程を飛ばして再開する
同じjob-1を再開します。chargeの完了行を見つけるため、課金の行を追加しません。
flow.run("job-1", ship_ok=True)完了した工程を確認して発送を再開する
flow.status()課金が重複していないことを読むcharge=already_completedとなり、shipが追加されます。金額100の行は1件のままです。
7もう一度再開しても保存結果を保つ
同じjobをさらに実行します。工程の主キーと決済の主キーが重複行を防ぎます。
flow.run("job-1", ship_ok=True)同じjobを再実行する
flow.status()完了行と決済行の数を確認する工程2行・決済1行のままです。ただし外部の決済APIへ置き換える場合、DB内のトランザクションだけでは原子性を保てないため、相手側の冪等キーなども必要になります。
まとめ
配信先ごとの完了を保持すると、一部の失敗だけを再処理できます。
工程の再開では完了済みの副作用を繰り返さない条件が必要です。
DB内の確定方法を、そのまま外部APIの原子性の保証にはできません。
操作履歴と解説を振り返れます。
Python 3.15 / SQLite / 固定時計のイベント処理モデル
from workflow import Delivery
flow = Delivery()
flow.publish("order-7")[('billing', 'order-7', 'pending'), ('mail', 'order-7', 'pending')]flow.consume("billing", "order-7")
flow.consume("mail", "order-7", fail=True)[('billing', 'order-7', 'done'), ('mail', 'order-7', 'pending')]
mail: failed; pending retainedflow.publish("order-7")[('billing', 'order-7', 'done'), ('mail', 'order-7', 'pending')]flow.consume("mail", "order-7")[('billing', 'order-7', 'done'), ('mail', 'order-7', 'done')]flow.run("job-1", ship_ok=False)
flow.status()charge=committed
ship=failed
steps=[('job-1', 'charge')]
charges=[('job-1', 100)]flow.run("job-1", ship_ok=True)
flow.status()charge=already_completed
ship=completed
steps=[('job-1', 'charge'), ('job-1', 'ship')]
charges=[('job-1', 100)]flow.run("job-1", ship_ok=True)
flow.status()charge=already_completed
ship=completed
steps=[('job-1', 'charge'), ('job-1', 'ship')]
charges=[('job-1', 100)]演習完了