お菓子の注文は保存されたのに倉庫は知らなかった

目標

業務と送信予定、受信IDと業務上の効果を、それぞれアトミックに保存し、実際のHTTP再送とプロセス終了のあとで復旧します。

なぜ重要なのか

応答がないという理由だけで業務が実行されていないと断定すると、重複した反映を生みます。逆に、送信の前に完了印を付けると、まだ配信していない業務を失います。PythonとSQLiteで2つの境界を分けて実装し、後始末のコードが実行されないプロセス終了でも残る状態を確認します。

予想の所要時間は85分です。基本のセッションより長いので、期限が切れる前に＋時間を押して延長してください。セッションが終了するとファイルが消えます。必要なコードは別に保管してください。Pythonの関数・例外・SQLiteトランザクションと、前のモジュールのACK復旧を理解してから始めます。

データ契約

提出するコードはすべて/root/outbox/worker.pyです。以下の関数が受け取るconは、正しい役割の開いた接続で、呼び出しの前にトランザクションはありません。open_storeを除く関数は、借りた接続を閉じません。成功・失敗のいずれのあとも、開いたトランザクションを残しません。sourceとsinkは必ず別々のファイルでテストし、本番DBには接続しません。

sourceのスキーマ:

CREATE TABLE orders (id TEXT PRIMARY KEY, qty INTEGER NOT NULL);
CREATE TABLE outbox (seq INTEGER PRIMARY KEY AUTOINCREMENT,
    id TEXT NOT NULL UNIQUE, qty INTEGER NOT NULL,
    sent INTEGER NOT NULL CHECK(sent IN (0,1)));

sinkのスキーマ:

CREATE TABLE inbox (id TEXT PRIMARY KEY, qty INTEGER NOT NULL);
CREATE TABLE stock (id INTEGER PRIMARY KEY CHECK(id=1), total INTEGER NOT NULL);

stockの初期行は(1,0)です。各役割の初期化は1つのトランザクションで行い、既存の内容を保存します。採点ツールが一時DB・loopback HTTPサーバー・子プロセスを作って後始末するので、DBのパスや固定ポートをコードに入れないでください。

ステップ

送信IDと数量の契約を作る: worker.pyに、Exceptionのサブクラスとして、ConflictとDeliveryError、そしてvalidate_event(event)を実装してください。eventは、idとqtyの2つのキーだけを持つdictです。idはASCIIの英数字・アンダースコア・ハイフンの1–64文字、qtyはboolを除くintの1–1000です。不正ならValueError、正常なら入力を変えずに、新しいdictのコピーを返します。
異なる2つのストアを初期化する: open_store(path, role)は、sourceまたはsinkの役割のsqlite3.Connectionを返します。isolation_level=None・busy timeout 1秒で開き、下のスキーマを、存在しないときにだけ作成してください。sinkのstockの初期行(1,0)は、存在しないときにだけ入れます。初期化は1つのトランザクションで行い、失敗したときは接続を閉じてエラーを伝えます。既存の記録を初期値で上書きしません。それ以外のroleは、接続の前にValueErrorです。
注文と送信予定を一緒に残す: enqueue(con,event,fault=None)は、validate_eventのあとBEGIN IMMEDIATEで処理します。新しいIDは、ordersの挿入→fault('after-order')→outbox(sent=0)の挿入→fault('after-outbox')→COMMIT→fault('after-commit')の順に進めて、Trueを返します。faultは、ある場合にだけ呼び出します。既存のID・同じqtyなら、変更せずにFalse、違うqtyならConflictです。コミット前のエラーはすべてロールバックして元のエラーを伝え、コミット後のエラーは、確定した記録を消しません。
完了していないバッチを順番に読む: pending(con,limit)は、sent=0のoutboxを、seqの昇順で最大limit個返します。各要素は、idとqtyだけを持つ新しいdictです。limitはboolを除くintの1–16で、それ以外の値はValueErrorです。照会は、DBも入力も変更しません。
重複受信の効果を1回に制限する: receive(con,event,fault=None)は、sinkの接続で動作します。検証のあと、BEGIN IMMEDIATEで、新しいIDのinboxの挿入→fault('after-inbox')→stock.totalにqtyを加算→fault('after-total')→COMMIT→fault('after-commit')の順に進め、Trueを返します。同じID・qtyならFalse、同じIDで違うqtyならConflictです。違うIDで同じqtyは、別々の業務です。コミット前のエラーは全体をロールバックし、コミット後のエラーは確定した状態を保存し、どちらも元のエラーを伝えます。
確認した項目にだけ完了印を付ける: mark_sent(con,event)は、検証したid・qtyと一致するsourceのoutboxのsentだけを1に変えます。存在しないIDはValueError、違うqtyはConflictで、失敗したときはDBの状態を変えません。新しい印ならTrue、すでに完了ならFalseです。ordersとoutboxの行は削除せず、トランザクションも残しません。
ACKのあとに完了印を付け、最初の失敗で止める: dispatch(con,send,limit=8,fault=None)は、pendingの有限なバッチだけを順番に処理します。sendにeventのコピーを渡し、正確なTrueのACKを受け取ったあとで、fault('after-delivery')とmark_sentを呼び出します。False・None・1・文字列はDeliveryErrorで、sendやfaultの例外は、そのまま伝えます。最初の失敗で止まり、それ以前の完了は保存します。戻り値は、今回完了印を付けた個数です。送信中は書き込みトランザクションを握らず、コールバックが引数を変えても、最初に選んだ項目だけに完了印を付けます。
HTTPと実際のプロセス再起動で復旧する: send_http(url,event,timeout=1)を実装してください。validate_eventのあとHTTP POSTでJSONを送り、ステータス200、1024バイト以下のJSON dictで、idとacceptedの正確に2つのキーを持ち、要求と同じid、accepted is Trueのときにだけ、Trueを返します。不正なACKはDeliveryError、HTTP・接続のエラーは伝え、応答は閉じます。timeoutはboolを除く正の有限なint/floatで5秒以下で、そうでなければValueErrorです。許可するURLは(http://127.0.0.1:포트/events만)だけで、プレースホルダーはポート番号です。ポートは明示的に1–65535を指定し、ユーザー情報・クエリ・fragmentは含められません。自動のプロキシとリダイレクトは使いません。最終検査は、受信のコミット後にパブリッシャーを実際に終了させ、新しいプロセスが同じIDを再送して、在庫を1回だけ増やすかを確認します。

参考

最終診断コマンド: python3 -B /opt/fixtures/outbox/check.py 8 /root/outbox/worker.py。前のステップでは、8を該当するステップの番号に置き換えます。提出ファイルは変更せず、一時的なストアで実行し、採点は12秒に制限します。
標準ライブラリだけを使い、インターネットも、インストールも、追加のcapabilityも必要ありません。提供されるHTTPサーバーはloopback専用で、認証・TLSのない教育用です。send_httpのtimeoutは、ソケットのブロッキング動作ごとの待機上限であり、リクエスト全体の総期限ではありません。
単一の送信ループの有限バッチと、同じディスクでのプロセス再起動を検証します。自動リトライのスケジューラー・複数ワーカーのclaim/lease・ディスクの消失・外部決済の原子性・大規模な性能は検証しません。
受信のコミット後に応答が消えても、既存の業務を消さないでください。同じID・内容でもう一度配信し、効果が1回だけかを、独立した接続で確認します。

送信IDと数量の契約を作る

worker.pyに、Exceptionのサブクラスとして、ConflictとDeliveryError、そしてvalidate_event(event)を実装してください。eventは、idとqtyの2つのキーだけを持つdictです。idはASCIIの英数字・アンダースコア・ハイフンの1–64文字、qtyはboolを除くintの1–1000です。不正ならValueError、正常なら入力を変えずに、新しいdictのコピーを返します。

int(True)が1であるという事実と、数量が有効であるという契約は別です。正規表現は文字列全体を検査してください。

異なる2つのストアを初期化する

open_store(path, role)は、sourceまたはsinkの役割のsqlite3.Connectionを返します。isolation_level=None・busy timeout 1秒で開き、下のスキーマを、存在しないときにだけ作成してください。sinkのstockの初期行(1,0)は、存在しないときにだけ入れます。初期化は1つのトランザクションで行い、失敗したときは接続を閉じてエラーを伝えます。既存の記録を初期値で上書きしません。それ以外のroleは、接続の前にValueErrorです。

CREATE TABLE IF NOT EXISTSとINSERT OR IGNOREの役割は違います。成功したときは、開いたトランザクションを残さないでください。

注文と送信予定を一緒に残す

enqueue(con,event,fault=None)は、validate_eventのあとBEGIN IMMEDIATEで処理します。新しいIDは、ordersの挿入→fault('after-order')→outbox(sent=0)の挿入→fault('after-outbox')→COMMIT→fault('after-commit')の順に進めて、Trueを返します。faultは、ある場合にだけ呼び出します。既存のID・同じqtyなら、変更せずにFalse、違うqtyならConflictです。コミット前のエラーはすべてロールバックして元のエラーを伝え、コミット後のエラーは、確定した記録を消しません。

2つのINSERTの間にCOMMITを入れると、プロセス終了のときに注文だけが残ります。faultのポイントを維持してこそ、途中の状態をテストできます。

完了していないバッチを順番に読む

pending(con,limit)は、sent=0のoutboxを、seqの昇順で最大limit個返します。各要素は、idとqtyだけを持つ新しいdictです。limitはboolを除くintの1–16で、それ以外の値はValueErrorです。照会は、DBも入力も変更しません。

IDのzがaより先に受け付けられることがあります。業務IDの辞書順ではなく、保存した挿入順を使ってください。

重複受信の効果を1回に制限する

receive(con,event,fault=None)は、sinkの接続で動作します。検証のあと、BEGIN IMMEDIATEで、新しいIDのinboxの挿入→fault('after-inbox')→stock.totalにqtyを加算→fault('after-total')→COMMIT→fault('after-commit')の順に進め、Trueを返します。同じID・qtyならFalse、同じIDで違うqtyならConflictです。違うIDで同じqtyは、別々の業務です。コミット前のエラーは全体をロールバックし、コミット後のエラーは確定した状態を保存し、どちらも元のエラーを伝えます。

受信記録と業務上の効果を、別々にコミットしないでください。Falseは失敗ではなく、今回は新しい効果を適用しなかった有効な重複です。

確認した項目にだけ完了印を付ける

mark_sent(con,event)は、検証したid・qtyと一致するsourceのoutboxのsentだけを1に変えます。存在しないIDはValueError、違うqtyはConflictで、失敗したときはDBの状態を変えません。新しい印ならTrue、すでに完了ならFalseです。ordersとoutboxの行は削除せず、トランザクションも残しません。

あとで突き合わせる業務と送信予定を保存します。WHERE idの条件なしで、バッチ全体を完了にしないでください。

ACKのあとに完了印を付け、最初の失敗で止める

dispatch(con,send,limit=8,fault=None)は、pendingの有限なバッチだけを順番に処理します。sendにeventのコピーを渡し、正確なTrueのACKを受け取ったあとで、fault('after-delivery')とmark_sentを呼び出します。False・None・1・文字列はDeliveryErrorで、sendやfaultの例外は、そのまま伝えます。最初の失敗で止まり、それ以前の完了は保存します。戻り値は、今回完了印を付けた個数です。送信中は書き込みトランザクションを握らず、コールバックが引数を変えても、最初に選んだ項目だけに完了印を付けます。

完了印を先に付けると、送信の前に終了したときに未完了の項目を失います。receiveのFalseを、HTTP ACKの失敗と混同しないでください。

HTTPと実際のプロセス再起動で復旧する

send_http(url,event,timeout=1)を実装してください。validate_eventのあとHTTP POSTでJSONを送り、ステータス200、1024バイト以下のJSON dictで、idとacceptedの正確に2つのキーを持ち、要求と同じid、accepted is Trueのときにだけ、Trueを返します。不正なACKはDeliveryError、HTTP・接続のエラーは伝え、応答は閉じます。timeoutはboolを除く正の有限なint/floatで5秒以下で、そうでなければValueErrorです。許可するURLは(http://127.0.0.1:포트/events만)だけで、プレースホルダーはポート番号です。ポートは明示的に1–65535を指定し、ユーザー情報・クエリ・fragmentは含められません。自動のプロキシとリダイレクトは使いません。最終検査は、受信のコミット後にパブリッシャーを実際に終了させ、新しいプロセスが同じIDを再送して、在庫を1回だけ増やすかを確認します。

urllib.requestのProxyHandler({})とHTTPRedirectHandlerを確認してください。リクエストが2回届いても、倉庫の効果は1回でなければなりません。