TT Lab
はじめる
学ぶ 学習パス コース

ACKの前に止まったお菓子の自販機

週末に止まった表示板と消えたお菓子の記録

TT Labで続きを見る

目標

週末のあいだ消えていたお菓子の在庫表示板を復旧します。ログの保管範囲が終わっていたら、同じ時点のスナップショットをインストールして、次のイベントから追いかけます。

なぜ重要なのか

接続が復旧しても、すでに消したイベントは戻ってきません。在庫とカーソルが別の時点を指していると、再生に成功しても誤った数字を表示します。今回は、保管境界・世代・同一時点の読み取り・アトミックなインストールを自分で実装し、実際の終了後のディスクの状態で確認します。

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

データ契約

成果物は/root/snapshot/worker.pyの1ファイルです。元データとレプリカは、同じ下のスキーマを持つ、別々の使い捨てのローカルSQLiteファイルです。採点ツールが作って後始末するので、DBのパスをハードコードしません。いずれも正常なスキーマの、信頼したローカルファイルで、関数に貸したconは、開始時に開いたトランザクションがありません。open_store・sync_files以外の関数は、借りた接続を閉じず、成功・失敗のあとにトランザクションを残しません。

CREATE TABLE checkpoint (id INTEGER PRIMARY KEY CHECK(id=1),
    epoch TEXT NOT NULL, last INTEGER NOT NULL, floor INTEGER NOT NULL);
CREATE TABLE stock (id INTEGER PRIMARY KEY CHECK(id=1), total INTEGER NOT NULL);
CREATE TABLE events (seq INTEGER PRIMARY KEY, delta INTEGER NOT NULL);

checkpointの初期行は(1, 渡したepoch, -1, 0)、stockの初期行は(1,0)です。同じ世代の中では番号を再利用せず、retained eventsはfloorからlastまで連続しています。すべて切り詰められた場合、floor=last+1です。totalは、初期値0から各deltaを足した業務状態です。expected_epochは、信頼した接続設定から受け取り、snapshot自体から取り出して期待値にはしません。

ラボの書き込みは、順番に1つずつ実行します。読み取りトランザクションの間に、別の接続が書き込みをコミットする方式で、時点の分離を検査します。別スレッドによる同時チェックポイントや、複数writerの性能テストではありません。すべての接続にwal_autocheckpoint=0を設定し、書き込みが終わったあとで順番に接続を閉じます。長期運用のWAL管理ポリシーとして、そのまま使わないでください。

ステップ

  1. 世代と復旧状態を保存する: worker.pyに、Exceptionのサブクラスとして、Conflict・ResyncRequired・Gap・CoveredBySnapshotを定義し、validate_snapshot(snapshot)とopen_store(path,epoch)を実装してください。スナップショットは、version・epoch・last・totalだけを持つdictです。versionは正確なint 1、epochはASCIIの英数字・アンダースコア・ハイフンの1–64文字、lastはboolを除くintで-1から2147483647、totalはboolを除くintで、絶対値は1000*(last+1)以下です。有効なら新しいdictのコピー、そうでなければValueErrorです。open_storeは、epochを検証したあと、下の3つのテーブルと初期行を、存在しないときにだけアトミックに作成します。isolation_level=None、timeout=1、WAL・synchronous=FULL・wal_autocheckpoint=0で開き、sqlite3.Connectionを返します。既存の内容は上書きせず、初期化に失敗したときは接続を閉じます。
  2. 在庫と次の番号を一緒に確定する: append(con,delta,fault=None)は、boolを除くintで-1000から1000の変化を追加します。BEGIN IMMEDIATEの中で、次のseq=last+1を検査し、eventsの挿入→fault('after-event')→stock.totalにdeltaを加算してcheckpoint.lastを更新→fault('after-state')→COMMIT→fault('after-commit')の順に進めます。seqは0から2147483647で、範囲外はValueErrorです。faultはある場合に呼び出し、seqを返します。コミット前のエラーは全体をロールバックし、コミット後のエラーは確定した状態を保存して、どちらも元のエラーを伝えます。
  3. 同一時点のスナップショットをエクスポートする: export_snapshot(con,between=None)は、BEGINの読み取りトランザクションで、epoch・lastの照会→between()→totalの照会→COMMITの順に、version=1のスナップショットを返します。フックは、ある場合にだけ1回呼び出し、別の接続の書き込みがここでコミットできなければなりません。失敗したときは読み取りトランザクションを後始末して、元のエラーを伝えます。2つのSELECTの間で、COMMITやBEGIN IMMEDIATEは使いません。
  4. 保管範囲と再生カーソルを比べる: trim(con,through)は、boolを除くintの-1から2147483647を受け取り、1つの書き込みトランザクションで、floor-1 <= through <= lastかどうかを検査します。そうでなければValueError、そうならseq<=throughの削除とfloor=through+1の更新だけを行い、Noneを返します。replay(con,epoch,last,limit=16)は、epochの形式、boolを除くlastの-1から2147483647、limitの1から16を検証します。同じ読み取りトランザクションで、別の世代はResyncRequired、サーバーより先に進んだカーソルはValueError、lastlastの(seq,delta)のtupleのlistを、番号順に最大limit個返します。空の結果と欠落エラーを区別します。
  5. スナップショットを一度にインストールする: install_snapshot(con,snapshot,expected_epoch,fault=None)は、snapshotとexpected_epochの形式を検証します。snapshot.epochと期待する世代が違えばResyncRequiredです。1つの書き込みトランザクションで、同じ世代のより小さいlast、または同じlastで違うtotalはConflict、同じlast・totalは変更せずにFalseです。それ以外は、eventsの全削除→fault('after-clear')→totalの置き換えとcheckpointのepoch・last・floor=last+1の更新→fault('after-install')→COMMIT→fault('after-commit')のあとでTrueを返します。コミット前のエラーは全体をロールバックし、コミット後のエラーは確定した状態を保存します。許可された別の世代は、小さな番号でも置き換えられます。
  6. 検証できる重複と新しいバッチを適用する: apply_batch(con,epoch,events,fault=None)は、epochとeventsを先に検証します。eventsは最大16個のlistで、空のlistも許可します。要素はseqとdeltaの2つからなるtuple/list、seqはboolを除くintで0から2147483647、deltaはboolを除くintで-1000から1000です。形式のエラーはValueError、バッチ内部の番号が1ずつ増えなければGapです。1つの書き込みトランザクションで、世代の不一致はResyncRequiredです。seq<=lastは、保管されている同じdeltaなら効果なし、違うdeltaはConflict、個々の履歴がなければCoveredBySnapshotです。新しいseqはlast+1だけを許可し、そうでなければGapです。新しいイベント・在庫・カーソルを更新するたびにfault('after-one')、全体のCOMMITのあとにfault('after-commit')を呼び出します。今回の新しい効果の数を返し、入力は変更しません。コミット前のエラーはバッチ全体をロールバックし、それ以後のエラーは確定した状態を保存します。
  7. 1回の復旧の試みを有限回で終える: sync_once(source,replica,expected_epoch,after_snapshot=None)は、期待する世代を検証し、sourceの実際の世代と違えばResyncRequiredを出します。replicaのepoch・lastでreplayを1回要求します。ResyncRequiredのときにだけ、export_snapshot(source)→install_snapshot(replica,...,expected_epoch)→after_snapshot()→スナップショットのlast以降のreplayで復旧します。フックは、ある場合にだけ呼び出します。バッチは最大16個で、apply_batchのあとにexport_snapshot(replica)を返します。2回目のreplayで再び切り詰められていたら、エラーをそのまま伝え、インストールした有効な状態を維持します。ほかのエラーも伝え、内部での無限リトライはしません。
  8. 実際の終了のあとに別々のファイルから復旧する: sync_files(source_path,replica_path,expected_epoch)は、期待する世代を検証し、ローカルの元ファイルがなければFileNotFoundErrorです。realpathが同じか、既存の2つのパスがsamefileなら、開く前にValueErrorです。別々の元データとレプリカをopen_storeで開き、sync_onceを1回実行した結果を返し、すべての経路で、自分が所有する接続を閉じます。レプリカを開けなくても、元データは閉じます。採点ツールは、append・install・batchのコミットの前後で、実際の子プロセスを終了コード73で終わらせ、新しい接続でディスクの状態を検査します。

参考

世代と復旧状態を保存する

worker.pyに、Exceptionのサブクラスとして、Conflict・ResyncRequired・Gap・CoveredBySnapshotを定義し、validate_snapshot(snapshot)とopen_store(path,epoch)を実装してください。スナップショットは、version・epoch・last・totalだけを持つdictです。versionは正確なint 1、epochはASCIIの英数字・アンダースコア・ハイフンの1–64文字、lastはboolを除くintで-1から2147483647、totalはboolを除くintで、絶対値は1000*(last+1)以下です。有効なら新しいdictのコピー、そうでなければValueErrorです。open_storeは、epochを検証したあと、下の3つのテーブルと初期行を、存在しないときにだけアトミックに作成します。isolation_level=None、timeout=1、WAL・synchronous=FULL・wal_autocheckpoint=0で開き、sqlite3.Connectionを返します。既存の内容は上書きせず、初期化に失敗したときは接続を閉じます。

last=-1ならtotalは0だけが可能です。CREATE IF NOT EXISTSと初期行の挿入を分け、すべてのint契約でboolを除いてください。

在庫と次の番号を一緒に確定する

append(con,delta,fault=None)は、boolを除くintで-1000から1000の変化を追加します。BEGIN IMMEDIATEの中で、次のseq=last+1を検査し、eventsの挿入→fault('after-event')→stock.totalにdeltaを加算してcheckpoint.lastを更新→fault('after-state')→COMMIT→fault('after-commit')の順に進めます。seqは0から2147483647で、範囲外はValueErrorです。faultはある場合に呼び出し、seqを返します。コミット前のエラーは全体をロールバックし、コミット後のエラーは確定した状態を保存して、どちらも元のエラーを伝えます。

ログを切り詰めても、番号はlastから続きます。MAX(events.seq)を、次の番号の根拠にしないでください。

同一時点のスナップショットをエクスポートする

export_snapshot(con,between=None)は、BEGINの読み取りトランザクションで、epoch・lastの照会→between()→totalの照会→COMMITの順に、version=1のスナップショットを返します。フックは、ある場合にだけ1回呼び出し、別の接続の書き込みがここでコミットできなければなりません。失敗したときは読み取りトランザクションを後始末して、元のエラーを伝えます。2つのSELECTの間で、COMMITやBEGIN IMMEDIATEは使いません。

1つの値ではなく、値の組み合わせが採点の対象です。フックの中で発生した新しい在庫は、次のexportで見えなければなりません。

保管範囲と再生カーソルを比べる

trim(con,through)は、boolを除くintの-1から2147483647を受け取り、1つの書き込みトランザクションで、floor-1 <= through <= lastかどうかを検査します。そうでなければValueError、そうならseq<=throughの削除とfloor=through+1の更新だけを行い、Noneを返します。replay(con,epoch,last,limit=16)は、epochの形式、boolを除くlastの-1から2147483647、limitの1から16を検証します。同じ読み取りトランザクションで、別の世代はResyncRequired、サーバーより先に進んだカーソルはValueError、lastlastの(seq,delta)のtupleのlistを、番号順に最大limit個返します。空の結果と欠落エラーを区別します。

すべて消したあとでも、last・floorは残る必要があります。不等号の等号1つで、すでに処理したイベントを再び含めてしまうことがあります。

スナップショットを一度にインストールする

install_snapshot(con,snapshot,expected_epoch,fault=None)は、snapshotとexpected_epochの形式を検証します。snapshot.epochと期待する世代が違えばResyncRequiredです。1つの書き込みトランザクションで、同じ世代のより小さいlast、または同じlastで違うtotalはConflict、同じlast・totalは変更せずにFalseです。それ以外は、eventsの全削除→fault('after-clear')→totalの置き換えとcheckpointのepoch・last・floor=last+1の更新→fault('after-install')→COMMIT→fault('after-commit')のあとでTrueを返します。コミット前のエラーは全体をロールバックし、コミット後のエラーは確定した状態を保存します。許可された別の世代は、小さな番号でも置き換えられます。

スナップショット自身に、許可された世代かどうかを尋ねないでください。expected_epochは、検証の対象の外から来た基準です。

検証できる重複と新しいバッチを適用する

apply_batch(con,epoch,events,fault=None)は、epochとeventsを先に検証します。eventsは最大16個のlistで、空のlistも許可します。要素はseqとdeltaの2つからなるtuple/list、seqはboolを除くintで0から2147483647、deltaはboolを除くintで-1000から1000です。形式のエラーはValueError、バッチ内部の番号が1ずつ増えなければGapです。1つの書き込みトランザクションで、世代の不一致はResyncRequiredです。seq<=lastは、保管されている同じdeltaなら効果なし、違うdeltaはConflict、個々の履歴がなければCoveredBySnapshotです。新しいseqはlast+1だけを許可し、そうでなければGapです。新しいイベント・在庫・カーソルを更新するたびにfault('after-one')、全体のCOMMITのあとにfault('after-commit')を呼び出します。今回の新しい効果の数を返し、入力は変更しません。コミット前のエラーはバッチ全体をロールバックし、それ以後のエラーは確定した状態を保存します。

入力全体を検証したあとで、小さな1つのバッチをアトミックに処理してください。スナップショットの合計から、個々の過去のdeltaを知ることはできません。

1回の復旧の試みを有限回で終える

sync_once(source,replica,expected_epoch,after_snapshot=None)は、期待する世代を検証し、sourceの実際の世代と違えばResyncRequiredを出します。replicaのepoch・lastでreplayを1回要求します。ResyncRequiredのときにだけ、export_snapshot(source)→install_snapshot(replica,...,expected_epoch)→after_snapshot()→スナップショットのlast以降のreplayで復旧します。フックは、ある場合にだけ呼び出します。バッチは最大16個で、apply_batchのあとにexport_snapshot(replica)を返します。2回目のreplayで再び切り詰められていたら、エラーをそのまま伝え、インストールした有効な状態を維持します。ほかのエラーも伝え、内部での無限リトライはしません。

最初の欠落は復旧経路を選ぶ合図で、復旧の途中の2回目の欠落は、今回の試みが終わっていない合図です。

実際の終了のあとに別々のファイルから復旧する

sync_files(source_path,replica_path,expected_epoch)は、期待する世代を検証し、ローカルの元ファイルがなければFileNotFoundErrorです。realpathが同じか、既存の2つのパスがsamefileなら、開く前にValueErrorです。別々の元データとレプリカをopen_storeで開き、sync_onceを1回実行した結果を返し、すべての経路で、自分が所有する接続を閉じます。レプリカを開けなくても、元データは閉じます。採点ツールは、append・install・batchのコミットの前後で、実際の子プロセスを終了コード73で終わらせ、新しい接続でディスクの状態を検査します。

with sqlite3.Connectionは、トランザクション管理と接続の終了が別です。開いた接続を誰が所有するのかを、はっきりさせてください。