メモAPIの履歴処理を別プロセスに分けてArchifyで追ってみた

前回は、メモと履歴を一つのトランザクションで保存しました。履歴が書けなければ、メモも戻して500を返す実装です。

次は保存後の処理を別プロセスへ渡した場合も見たいと書いたので、今回は履歴を作るワーカーを追加します。私が確認したいのは、APIが成功を返す時点と、履歴処理が終わる時点を、図でも区別できるかです。

前回の実装は残し、examples/archify-notes-api-outboxに今回のコードを用意しました。引き続きPython標準ライブラリとSQLiteだけを使います。

変更したのは、APIのトランザクションで保存する対象です。メモと履歴の組み合わせを、メモと未処理ジョブの組み合わせに変えました。

前回：notes ＋ note_events → COMMIT → HTTP 201
今回：notes ＋ jobs        → COMMIT → HTTP 202

ジョブは、後で履歴を作るための依頼です。保存するのはメモのIDと状態で、最初はpendingにします。

メモを保存してからメモリ上のキューへ渡す方法だと、その間にAPIが止まった場合を別に考える必要があります。今回はメモと依頼を一緒に残したかったので、同じSQLiteファイルにjobsテーブルを追加しました。

業務データと後続処理の依頼を同じトランザクションで書く部分は、transactional outboxの考え方を参考にしています。ただし今回は外部のメッセージブローカーへ配信しません。同じDB内の履歴を作るための、小さな検証用の実装です。AWSのパターン解説

API側の保存処理は、要点を抜き出すと次の形です。

with connection:
    connection.execute("BEGIN")
    cursor = connection.execute(
        "INSERT INTO notes(text) VALUES (?)", (text,)
    )
    note_id = cursor.lastrowid
    job = connection.execute(
        "INSERT INTO jobs(note_id, status) VALUES (?, ?)",
        (note_id, "pending"),
    )
    job_id = job.lastrowid
# メモとジョブを確定してからHTTP応答へ進む

jobsのINSERTが失敗すれば、同じトランザクション内の新しいメモも戻ります。ここは前回の保存処理と同じ考え方です。履歴を書き込む箇所だけを、APIから外しました。

追加したファイルはnotes_api/worker.pyです。起動するたびに、未処理ジョブを最大1件読みます。

with connection:
    connection.execute("BEGIN IMMEDIATE")
    job = connection.execute(
        "SELECT id, note_id FROM jobs WHERE status = 'pending' "
        "ORDER BY id LIMIT 1"
    ).fetchone()
    if job is not None:
        job_id, note_id = job
        connection.execute(
            "INSERT INTO note_events(note_id, kind) VALUES (?, ?)",
            (note_id, "created"),
        )
        connection.execute(
            "UPDATE jobs SET status = 'completed' WHERE id = ?",
            (job_id,),
        )

履歴のINSERTと、ジョブを完了にするUPDATEは一緒に確定します。API側のコミットとは別のトランザクションです。with connection:を使う点は前回と同じで、接続を閉じる処理は外側のclosing()に分けています。Python公式ドキュメント

今回は常駐して何度もポーリングする処理にはしませんでした。ワーカーを止めた状態を作り、APIだけ動いているときのDBを見たかったためです。自動再試行も入れていません。失敗時はジョブをpendingのまま残し、終了コード1で止まります。

最初に整理しておきたかったのが、この二つのコミットでした。同じDBファイルを使っていても、ワーカー側の失敗で、すでに確定したAPI側のメモまで戻るわけではありません。

今回のPOST /notesは202を返します。メモ自体は保存済みですが、履歴を作る処理まで完了したとは伝えないことにしました。

HTTPの202は、処理を受け付けたものの、まだ完了していない場合の応答です。後続処理の成功まで約束するステータスではありません。RFC 9110の202 Accepted

メモの作成だけをAPIの契約にするなら、201を返す設計も考えられます。今回は履歴処理まで含む依頼の受付を表すために202を選び、確認先を応答に入れました。

{
  "id": 1,
  "text": "別プロセスで履歴保存",
  "job_id": 1,
  "status_url": "/jobs/1"
}

追加したGET /jobs/{id}は、ジョブの現在状態を返します。ワーカーを動かしていない時点では、次の応答でした。

{"id": 1, "note_id": 1, "status": "pending"}

このGETのHTTPステータスは200です。状態を読めたことと、ジョブが完了したことを混同しないようにしました。完了したかどうかは、本文のstatusで見ます。

POSTの応答にはpendingを固定で入れていません。ワーカーを同時に動かす構成なら、HTTP応答を返すまでに処理が終わる可能性もあるためです。今回は意図的にワーカーを止め、202を確認してから起動しました。

前回の図に追加する箱は、履歴ワーカーです。APIとは別プロセスで起動するため、APIの枠の外に置きました。

今回のコードには、APIからワーカーを呼ぶ処理がありません。APIはDBにジョブを残し、ワーカーは起動後にDBを読みます。そのため、APIからワーカーへ直接矢印をつなぐと、実装とは違う関係になります。

独立したキュー製品の箱も追加していません。ジョブの保存先は同じSQLiteファイルなので、DBの説明をメモ・ジョブ・履歴の3表に変えました。

前回と今回のJSONをArchifyのcompare architectureで比較すると、箱の追加が1件、接続の追加も1件でした。ほかに保存対象などのラベルも変わっています。

生成した図は9項目の検証を通り、4種類の画面サイズでもはみ出しはありませんでした。固定のViewer UIとhtml langは英語ですが、図の説明は日本語で記載しています。

実装ディレクトリで、APIを起動します。ローカル検証用なので、待ち受け先は127.0.0.1です。

python3 -m notes_api.server --db notes.sqlite3 --port 8765

別のターミナルからメモを登録し、状態を確認します。以下は新しいDBで、最初のIDが1の場合です。

curl -i http://127.0.0.1:8765/notes \
  -H 'Content-Type: application/json' \
  -d '{"text":"別プロセスで履歴保存"}'
curl http://127.0.0.1:8765/jobs/1
curl http://127.0.0.1:8765/notes/1

実際の記録では、POSTは202でした。メモはGETで取得できますが、ジョブはpendingで、履歴テーブルは0行です。

ここでAPIのプロセスを終了し、同じDBで起動し直しました。それでもメモと未処理ジョブは残っています。少なくとも今回の依頼は、APIプロセスのメモリだけに置かれていません。

次に、APIと同じディレクトリの別ターミナルでワーカーを動かします。同じDBファイルを指定する点はそろえておきます。

python3 -m notes_api.worker --db notes.sqlite3
curl http://127.0.0.1:8765/jobs/1

正常に処理できると、履歴が1行増え、ジョブがcompletedになります。APIの応答を後から書き換えるのではなく、別のGETで完了を確認する形です。

前回と比べるために、今回もSQLiteのトリガーで履歴のINSERTを意図的に失敗させました。検証用の一時DBだけに、次のトリガーを作っています。

CREATE TRIGGER fail_audit
BEFORE INSERT ON note_events
BEGIN
    SELECT RAISE(ABORT, 'injected audit failure');
END;

RAISE(ABORT, ...)で履歴のINSERTを中断し、例外を受けたPython側のコンテキストマネージャーがワーカーのトランザクションを戻します。SQLite公式ドキュメント

ワーカーは終了コード1で止まりました。別接続からDBをSELECTすると、メモ1行、履歴0行、ジョブはpendingでした。前回のようにメモまで戻ることはありません。

APIはすでに202を返しています。ここでワーカーが失敗しても、過去のHTTP応答が500に変わるわけではありません。失敗を確認する場所が、リクエストの応答からワーカーの実行記録へ移ったと感じました。

トリガーを削除し、新しいプロセスでワーカーをもう一度起動すると、履歴が作られてcompletedになりました。これは原因を除いた後の手動再実行です。自動的にリトライして復旧した結果ではありません。

対照として、API側のジョブ登録も失敗させました。この場合はPOSTが500になり、新しいメモもジョブも残りません。保存前の失敗と、受付後の失敗では、戻す範囲が違います。

テストと実行記録は、次のコマンドで再現できます。

python3 -m unittest discover -s tests -v
python3 scripts/verify_runtime.py

16件のテストが通りました。別の実行スクリプトでは、9つのHTTPケースと、ワーカーを別プロセスで3回起動した結果を保存しています。APIも2つのプロセスを使い、終了・再起動を挟みました。環境はPython 3.13.5、SQLite 3.51.2です。

履歴を書いた後の完了UPDATEも、別のテストで失敗させています。その場合は履歴も戻り、受付済みのメモとpendingジョブが保持されました。ワーカー内でも、履歴だけを先に確定しないことを確認しています。

別プロセスにしたことで処理が速くなった、という検証でもありません。ワーカーのBEGIN IMMEDIATEは書き込みトランザクションを開始するので、同じSQLiteへのAPIの書き込みが待つ場合があります。今回は外部通信などの長い処理を、その中に入れていません。SQLiteのトランザクション仕様

複数ワーカーの同時実行、処理中の強制終了、クライアントによるPOSTの再送はまだ試していません。同じPOSTを同じメモにまとめる仕組みもないため、今回の実装で重複が起きないとは言えません。

今回は、図にワーカーの箱を追加するだけでなく、202が示す範囲と、失敗時に戻る範囲を更新しました。APIが応答できることと、後続処理が進んでいることは分けて見ないといけないと感じました。GET /healthも、ワーカーの稼働状況までは確認しません。

ここまで読んでいただき、ありがとうございます。もしこの記事の技術や考え方に少しでも興味を持っていただけたら、ネクストのエンジニアと気軽に話してみませんか。

• 選考ではありません
• 履歴書不要
• 技術の話が中心
• 所要時間30分程度
• オンラインOK

エンジニアと話してみる