レプリケーションとフェイルオーバー
主系・複製先を別々のSQLiteファイルで用意し、連番付きの業務ログを転送する構成を使います。非同期の成功応答、適用の遅れ、複製完了を待つ応答、主系切替後の値を順に確認します。これはアプリケーションがログを転送する方式で、各DB製品のネイティブ複製プロトコルとは別の構成です。
説明
主系で変更が確定した時点と、複製先が同じ変更を受け取り反映した時点には差が生じます。どの時点で利用者へ成功を返すかによって、待ち時間と障害時に失う可能性のある変更が変わります。
主系・複製先を別々のSQLiteファイルで用意し、連番付きの業務ログを転送する構成を使います。非同期の成功応答、適用の遅れ、複製完了を待つ応答、主系切替後の値を順に確認します。これはアプリケーションがログを転送する方式で、各DB製品のネイティブ複製プロトコルとは別の構成です。
1別ファイルに主系と複製先を用意する
どちらも残高100、適用位置0から始めます。
import sqlite3,tempfile
from pathlib import Path
from replication_tools import initialize,append,replicate,state,position
tmp = tempfile.TemporaryDirectory()
primary = sqlite3.connect(Path(tmp.name)/'primary.db'主系の保存ファイル,isolation_level=None)
replica = sqlite3.connect(Path(tmp.name)/'replica.db'複製先の保存ファイル,isolation_level=None)
initialize(primary)
initialize(replica)
print('primary:',state(primary)主系の値と位置)
print('replica:',state(replica))positionは適用済みの最後の変更番号です。appendは残高変更とログ保存を一つのトランザクションにし、replicateは未適用の番号を順に複製先へ適用します。各側の値と位置を合わせて確認します。
2主系の確定だけで成功を返す
番号1で20を加算し、複製前に成功位置を記録します。
append(primary,1,20)主系へ変更1を確定
acknowledged = 1複製前に成功応答を確定
print("acknowledged:",acknowledged)
print("primary:",state(primary))
print("replica:",state(replica))主系は位置1・120、複製先は位置0・100です。この応答方針では、複製を待たずに成功を返すため、成功済みでも複製先へまだ届いていない期間があります。
3ログを複製先へ適用する
未適用の変更を転送します。
print("applied:",replicate(primary,replica)不足する連番を順に適用)
print("replica:",state(replica)反映済みの値)位置1・120へ追いつきます。この構成のreplicateは受信と適用を続けて行います。製品によっては、受信・永続化・適用が別々の位置として管理されるため、監視する位置の意味も確認します。
4成功応答を返した変更のレプリケーションが遅れる
番号2で30を追加し、今回も主系だけで成功とします。
append(primary,2,30)変更2を主系へ確定
acknowledged = 2
print("primary:",state(primary))
print("replica:",state(replica))
print("acknowledged but unapplied:",acknowledged-position(replica)成功済みと適用済みの位置差)主系150、複製先120で、成功応答済みの一件が未反映です。この数字はこの例のログ件数差であり、時間の遅延そのものではありません。時間差を知るには時刻など別の情報が必要です。
5この時点で複製先に反映されていないデータ
現時点の複製先を切替先とした場合の残高を取得します。
print("candidate after failover:",state(replica)切替先候補が持つ状態)
print("missing successful sequence:",list(range(position(replica)+1,acknowledged+1)成功済みだが未反映の番号))切替先の残高は120で、成功した番号2の+30がありません。主系の保存データが回収できなければ、この成功済み変更を復元できない可能性があります。非同期の応答と、切替先が保持する範囲の差です。
6同期の対象を決めて成功を返す
まだ主系を利用できる状態で番号2を転送し、番号3は複製適用まで待ちます。
replicate(primary,replica)
append(primary,3,40)次の変更を主系へ確定
replicate(primary,replica)
acknowledged = 3 if position(replica)>=3複製先の適用完了を成功条件にする else acknowledged
print("acknowledged:",acknowledged,"replica:",state(replica))両側が190となってから番号3の成功を返します。ここでは複製先での適用完了を待つ方針です。同期複製という名前だけでは、受信・ディスクへの保存・適用のどこまで待つかは決まらないため、成功条件を具体化します。
7複製が止まると成功応答も進まない
番号4を主系へ追加し、転送をまだ行いません。
append(primary,4,10)主系では変更4を確定
can_ack = position(replica)>=4待っている位置へ届いたか
print("primary:",state(primary),"replica:",state(replica))
print("can acknowledge 4:",can_ack,"acknowledged:",acknowledged)主系200・複製先190、成功を返せるかはFalseです。複製先の適用を待つ分だけ、遅延や停止が応答へ影響します。安全な範囲を保つ条件と、処理を継続できる条件を同時に設計します。
8追いついてから成功を返す
番号4を適用します。
replicate(primary,replica)待っていた変更を適用
acknowledged = 4条件を満たした後に成功
print("acknowledged:",acknowledged,"replica:",state(replica))複製先も位置4・200になりました。この範囲で主系を失っても、ここまで成功を返した四件は複製先にあります。複製先自体の障害や破損まで同じ確認で保証するものではなく、保存先の独立性とバックアップも別に必要です。
9旧主系を停止して複製先で新しい変更を受ける
旧主系の接続を閉じ、新しい主系側へ番号5を追加します。
primary.close()旧主系の書込み経路を終了
append(replica,5,-20)切替先で新しい変更を確定
print("new primary:",state(replica))新しい主系は位置5・180です。複数の主系が同時に別の変更を受ける状態を避けるため、旧主系への書込みを停止してから受け付けます。実際の切替では、接続先の変更だけでなく、旧主系を確実に書込み不能にする制御も必要です。
10復旧した旧主系のデータを確認する
旧ファイルを読取り専用で開いて位置を確認します。
old_uri = (Path(tmp.name)/'primary.db').as_uri()+'?mode=ro'旧側を読取り専用で確認
old = sqlite3.connect(old_uri,uri=True)
print('old:',state(old)旧側の古い位置)
print('new:',state(replica)新側で進んだ位置)旧側は位置4・200、新側は位置5・180です。旧側が起動できることは、最新の主系として正しいことを意味しません。どちらの履歴を正本とするか決め、再同期してから役割を戻します。
まとめ
成功応答の条件と複製の受信・保存・適用位置を対応させます。
非同期では成功済みの未反映変更、同期では複製遅延による待ちを考慮します。
主系切替では旧主系の停止と復帰時の再同期まで含めて設計します。
操作履歴と解説を振り返れます。
Python / SQLite
import sqlite3,tempfile
from pathlib import Path
from replication_tools import initialize,append,replicate,state,position
tmp = tempfile.TemporaryDirectory()
primary = sqlite3.connect(Path(tmp.name)/'primary.db',isolation_level=None)
replica = sqlite3.connect(Path(tmp.name)/'replica.db',isolation_level=None)
initialize(primary)
initialize(replica)
print('primary:',state(primary))
print('replica:',state(replica))primary: {'position': 0, 'amount': 100}
replica: {'position': 0, 'amount': 100}append(primary,1,20)
acknowledged = 1
print("acknowledged:",acknowledged)
print("primary:",state(primary))
print("replica:",state(replica))acknowledged: 1
primary: {'position': 1, 'amount': 120}
replica: {'position': 0, 'amount': 100}print("applied:",replicate(primary,replica))
print("replica:",state(replica))applied: 1
replica: {'position': 1, 'amount': 120}append(primary,2,30)
acknowledged = 2
print("primary:",state(primary))
print("replica:",state(replica))
print("acknowledged but unapplied:",acknowledged-position(replica))primary: {'position': 2, 'amount': 150}
replica: {'position': 1, 'amount': 120}
acknowledged but unapplied: 1print("candidate after failover:",state(replica))
print("missing successful sequence:",list(range(position(replica)+1,acknowledged+1)))candidate after failover: {'position': 1, 'amount': 120}
missing successful sequence: [2]replicate(primary,replica)
append(primary,3,40)
replicate(primary,replica)
acknowledged = 3 if position(replica)>=3 else acknowledged
print("acknowledged:",acknowledged,"replica:",state(replica))acknowledged: 3 replica: {'position': 3, 'amount': 190}append(primary,4,10)
can_ack = position(replica)>=4
print("primary:",state(primary),"replica:",state(replica))
print("can acknowledge 4:",can_ack,"acknowledged:",acknowledged)primary: {'position': 4, 'amount': 200} replica: {'position': 3, 'amount': 190}
can acknowledge 4: False acknowledged: 3replicate(primary,replica)
acknowledged = 4
print("acknowledged:",acknowledged,"replica:",state(replica))acknowledged: 4 replica: {'position': 4, 'amount': 200}primary.close()
append(replica,5,-20)
print("new primary:",state(replica))new primary: {'position': 5, 'amount': 180}old_uri = (Path(tmp.name)/'primary.db').as_uri()+'?mode=ro'
old = sqlite3.connect(old_uri,uri=True)
print('old:',state(old))
print('new:',state(replica))old: {'position': 4, 'amount': 200}
new: {'position': 5, 'amount': 180}演習完了