イベント再配信でタスクが2件に？SQLiteで二重処理の起きる条件を検証

タスクはあるのに、完了記録がない

イベントを受け取るたびに、処理済みか確認する。未処理ならタスクを保存し、最後にイベントIDを記録する。一見すると、同じイベントをもう一度受け取っても大丈夫そうです。

ところが、タスクをコミットした直後に止まると、データベースにはタスクだけが残ります。再配信を受けたワーカーには完了記録が見つからず、もう1件追加してしまいます。

if not processed(event_id):
    insert_follow_up_job(event_id, job_key)
    commit()  # タスクはここで保存済み
    # この位置で止まると、完了記録はまだない
    insert_processed_event(event_id)
    commit()

これは処理の順序を示す疑似コードです。問題は、完了記録の有無だけでなく、タスクと完了記録のどちらまで保存されたかにあります。

同じイベントが届くこと自体は、異常な前提ではありません。RabbitMQの信頼性ガイドも、障害時の再配信に備え、繰り返し受け取っても同じ結果になる処理を推奨しています。この性質を冪等性と呼びます。以下ではRabbitMQを使わず、同じ入力を別のプロセスへ2回渡して、この停止位置だけを調べます。

停止位置を変えると、保存された行数はどう変わるか

架空のイベントIDを agent-run-042:completed、作るタスクのキーを review-agent-run-042 に固定しました。follow_up_jobs にタスク、processed_events に処理済みのIDを保存します。後者のIDは PRIMARY KEY NOT NULL です。比較用のタスク表には、イベントIDやタスクのキーへの一意制約を付けていません。

1回目のワーカーを決めた位置で os._exit(17) により終了させ、その後で別のプロセスへ同じイベントを渡しました。終了コード0を処理完了の模擬ACK、17をACKなしとして記録しています。正常条件でも2回渡し、通常の重複配信で1件に収まることを確認しました。

「別コミット・タスク確定後に停止」の条件では、最後のタスク数が2件になりました。処理済みIDの主キーは重複を防いでいます。それでも、1回目のタスクはすでに別のコミットで確定していたため、2回目のタスク追加を防げませんでした。

同一トランザクションでは、コミット前に止めるとタスク・処理済み記録ともに0件、コミット後に止めるとともに1件でした。どちらの停止状態から再配信しても、最終的なタスクは1件です。表は各条件1回の実行で得た保存行数で、成功率や性能の比較ではありません。

検証は2026年10月24日JSTに、AIアシスタントがLinuxサーバーのCPUで実行しました。Python 3.11.2、SQLite 3.40.1、journal_mode=DELETE、synchronous=FULL を使っています。LLM APIや実ブローカーは呼び出していません。

タスクと完了記録を一つのトランザクションに入れる

同じSQLiteファイルに両方を保存できるなら、まとめてコミットする形にします。今回のワーカーの中心部分は次のコードです。con は isolation_level=None で開いた接続で、トランザクションを明示して操作しています。

con.execute('BEGIN IMMEDIATE')
inserted = con.execute(
    'INSERT OR IGNORE INTO processed_events(event_id) VALUES (?)',
    (event_id,),
).rowcount
if inserted:
    con.execute(
        'INSERT INTO follow_up_jobs(event_id,job_key) VALUES (?,?)',
        (event_id, job_key),
    )
con.execute('COMMIT')
# この後で処理完了を返す

初めて受け取ったIDなら、完了記録とタスクを両方追加します。すでに記録済みなら挿入件数が0になり、タスクの追加も行いません。完了記録を先にINSERTしていても、COMMIT前には確定していないため、そこでプロセスが終了した試行ではタスクとともに残りませんでした。

SQLiteのトランザクション資料では、BEGIN IMMEDIATE は書き込みトランザクションを直ちに開始します。別の接続が書き込み中なら SQLITE_BUSY になる場合があります。今回はワーカーを順番に起動しており、並行実行の競合や再試行方針は測っていません。

ここで揃えたいのは、後続タスクを保存することと、そのイベントを処理済みにすることの確定タイミングです。一つのDB内でまとめられる処理を、途中まで確定した状態で残さないようにします。

外部APIと、別のIDで届く同じ仕事は別に考える

今回のタスクは、同じSQLiteファイルに入る1行です。メール送信や外部サービスへのタスク作成は、このトランザクションには入りません。DBの変更をロールバックしても、送信済みのメールやHTTPリクエストが取り消されるわけではありません。

外部APIを呼ぶなら、提供側の冪等キー、処理結果の照会、再試行時の扱いを確認します。DB内に送信待ちの仕事を保存する方式でも、その後に外部へ送る処理には重複への対策が必要です。この記事のコードを入れただけで、外部への操作まで1回にできるとは言えません。

また、今回の判定キーはイベントIDです。同じ仕事が別のイベントIDで届けば、2回とも新規として扱われます。業務上の同一操作を何で識別するか、記録をいつまで残すかは、別途決める必要があります。

タスク表へ適切な一意制約を設ける方法もありますが、制約に違反したときに完了記録やACKをどう扱うかまで含めて設計します。ここでは、その比較や複数ワーカー、ネットワーク障害、OS停止・電源断を検証していません。

処理を繰り返してもよいかという判断と、その処理を実行する権限があるかという判断も別です。許可の扱いはkiuribotsの権限設計と本番導入前の確認点で説明しています。

同じ4条件をPythonで再実行する

検証用ファイル一式（ZIP）を展開し、run-experiment.py と input.json のあるフォルダで実行します。Pythonの標準ライブラリだけを使い、APIキーや追加パッケージは不要です。

python run-experiment.py --out results-local

環境によっては python を python3 に置き換えてください。--out には存在しない出力フォルダを指定します。既存の結果を上書きしないよう、同じ名前のフォルダがある場合は停止します。

結果は results-local/results.json、各配信後に残った行は observations.jsonl、子プロセスの出力と終了コードは raw-processes.jsonl に保存されます。期待した行数や終了コードと一致しなければ、検証スクリプトはエラーで終了します。検証用のSQLiteファイルは一時フォルダに作り、終了後に削除します。

公開時の全体結果と保存行の記録、子プロセスの生ログも読めます。結果JSONに実行時刻、環境、コードと入力のSHA-256を残しました。時刻や環境の異なる再実行では、JSON全体が同じになるとは限りません。

自分の処理に置き換えるときは、まず「どの保存が終わった後に止まると、再配信で同じ仕事が増えるか」を一つ選んで確かめてください。処理済み記録とタスクが同じコミットに入っているかを、停止後に残ったデータから確認できます。