第169回では、移行前後のSQLiteへ同じ検索を実行し、列名、行順、値、型を突き合わせました。しかし、検証後に候補が変わったり、切り替え時に別のファイルを指定したりすれば、その試験結果を本番へ引き継げません。
今回は、現行データベースと検証済み候補のSHA-256を照合し、現行ファイルを退避してから、候補を稼働用の名前へ原子的に置き換えます。置き換え後にはハッシュとSQLiteの検査をやり直し、JSONレポートを保存します。
対象は、すべてのSQLite接続を閉じたローカルPOSIX環境です。このスクリプトはアプリケーションの停止、接続数の確認、再起動、業務上のヘルスチェックを行いません。
今回の切り替え契約
| 項目 | 仕様 |
|---|---|
| 動作環境 | Python 3.10以上、ローカルのPOSIXファイルシステム、外部ライブラリ不要 |
| 対象形式 | データベースヘッダーの読み書きバージョンがともに1である、WALではないSQLiteメインファイル |
| 現行ファイル | アプリ停止後のSQLiteメインファイル。指定したSHA-256と一致すること |
| 候補ファイル | 事前検査後に記録したSHA-256と一致すること |
| 配置 | 現行、候補、退避コピー、レポートを同じディレクトリに置く |
| 副ファイル | -wal、-shm、-journalが一つでもあれば停止する |
| ファイル種別 | 現行と候補は通常ファイルとし、シンボリックリンク、同じinode、既存の別ハードリンクを拒否する |
| 権限 | 現行と候補の所有者、グループ、モードが一致し、現行ファイルの所有者として実行する |
| 候補の検査 | integrity_check(1)、foreign_key_check、user_versionを確認する |
| 直前退避 | 現行ファイルを新しいパスへモード0600でコピーする。既存ファイルは上書きしない |
| 置き換え | os.replace(candidate, live)を1回実行する。成功すると候補の元パスはなくなる |
| 完了条件 | 稼働用パスの再検査、JSONレポートの保存、標準出力への成功JSONの書き込みとflushがすべて完了すること |
SHA-256はファイルの取り違えや意図しない変更を検出するための値です。期待値も同じ権限で書き換えられる環境では、改ざんの証明にはなりません。
WAL形式は接続前に拒否する
副ファイルが見当たらないだけでは、WALモードではないとは判断できません。副ファイルのないWAL形式のデータベースをmode=roで開いても、ディレクトリが書き込み可能ならSQLiteが-walや-shmを作成する場合があります。
そこで、この実装はSQLiteへ接続する前にデータベースヘッダーを読み、オフセット18の書き込みバージョンとオフセット19の読み取りバージョンがともに1であることを確認します。値2はWAL形式なので拒否します。スクリプト自身はWALからロールバックジャーナル方式への変換や、副ファイルの削除を行いません。
また、副ファイルがないことは接続数がゼロである証明にはなりません。アプリケーション固有の停止手順ですべての接続を閉じてから実行します。通常のファイルコピーを退避に使えるのも、トランザクションが進行していないことを前提にしているためです。
原子的なのはパスの置き換えだけ
os.replace()による同一ファイルシステム上の置き換えは、POSIXでは成功時に原子的です。この例は全パスを同じディレクトリに限定します。
アプリ停止、退避コピー、置き換え後検査、レポート作成までが一つのトランザクションになるわけではありません。置き換え後の処理に失敗しても、自動的には退避コピーへ戻しません。
切り替えスクリプト
次のコードをsqlite_cutover.pyとして保存します。SQLite接続はmode=roで行いますが、その前にWAL形式をヘッダーから拒否します。
from __future__ import annotations
import argparse
import hashlib
import json
import os
import shutil
import sqlite3
import stat
import sys
from contextlib import closing
from datetime import datetime, timezone
from pathlib import Path
class ContractError(Exception):
pass
class PostCutoverError(Exception):
pass
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open('rb') as source:
while chunk := source.read(1024 * 1024):
digest.update(chunk)
return digest.hexdigest()
def normalize(value: str) -> Path:
path = Path(os.path.abspath(os.path.expanduser(value)))
return path.parent.resolve(strict=True) / path.name
def lexists(path: Path) -> bool:
return os.path.lexists(os.fspath(path))
def require_regular(path: Path) -> os.stat_result:
try:
result = path.lstat()
except FileNotFoundError as exc:
raise ContractError(f'missing file: {path.name}') from exc
if stat.S_ISLNK(result.st_mode) or not stat.S_ISREG(result.st_mode):
raise ContractError(f'not a regular non-symlink file: {path.name}')
return result
def reject_sidecars(*databases: Path) -> None:
suffixes = ('-wal', '-shm', '-journal')
found = [
Path(str(database) + suffix).name
for database in databases
for suffix in suffixes
if lexists(Path(str(database) + suffix))
]
if found:
raise ContractError('SQLite sidecar exists: ' + ','.join(found))
def require_rollback_format(path: Path) -> None:
with path.open('rb') as source:
header = source.read(20)
if len(header) != 20 or header[:16] != b'SQLite format 3\x00':
raise ContractError(f'not a SQLite 3 database: {path.name}')
write_version = header[18]
read_version = header[19]
if write_version != 1 or read_version != 1:
raise ContractError(
f'WAL or unsupported file format: {path.name} '
f'(write={write_version}, read={read_version})'
)
def inspect_database(path: Path, expected_version: int) -> dict[str, object]:
uri = path.as_uri() + '?mode=ro'
with closing(sqlite3.connect(uri, uri=True, timeout=1.0)) as connection:
connection.execute('PRAGMA query_only = ON')
integrity = connection.execute(
'PRAGMA integrity_check(1)'
).fetchall()
foreign_key_error = connection.execute(
'PRAGMA foreign_key_check'
).fetchone()
version = connection.execute('PRAGMA user_version').fetchone()[0]
journal_mode = connection.execute(
'PRAGMA journal_mode'
).fetchone()[0]
if integrity != [('ok',)]:
raise ContractError('integrity_check did not return ok')
if foreign_key_error is not None:
raise ContractError('foreign_key_check found a violation')
if version != expected_version:
raise ContractError(
f'user_version mismatch: expected {expected_version}, got {version}'
)
if str(journal_mode).lower() == 'wal':
raise ContractError('WAL journal mode is not supported')
return {'user_version': version, 'journal_mode': journal_mode}
def create_exclusive_0600(path: str, flags: int) -> int:
descriptor = os.open(path, flags, 0o600)
try:
os.fchmod(descriptor, 0o600)
except BaseException:
os.close(descriptor)
raise
return descriptor
def copy_exclusive(source: Path, destination: Path) -> None:
with source.open('rb') as input_file, open(
destination,
'xb',
opener=create_exclusive_0600,
) as output_file:
shutil.copyfileobj(input_file, output_file, length=1024 * 1024)
output_file.flush()
os.fsync(output_file.fileno())
def write_report(path: Path, payload: dict[str, object]) -> None:
with open(
path,
'x',
encoding='utf-8',
newline='\n',
opener=create_exclusive_0600,
) as output:
json.dump(
payload,
output,
ensure_ascii=False,
separators=(',', ':'),
)
output.write('\n')
output.flush()
os.fsync(output.fileno())
def run(args: argparse.Namespace) -> dict[str, object]:
if os.name != 'posix':
raise ContractError('this implementation requires a POSIX system')
if args.expected_user_version < 0:
raise ContractError('expected-user-version must be zero or greater')
live = normalize(args.live)
candidate = normalize(args.candidate)
backup = normalize(args.backup)
report = normalize(args.report)
paths = (live, candidate, backup, report)
if len({str(path) for path in paths}) != 4:
raise ContractError('all four paths must be different')
if any(path.parent != live.parent for path in paths[1:]):
raise ContractError('all four paths must be in the same directory')
if lexists(backup) or lexists(report):
raise ContractError('backup or report already exists')
live_stat = require_regular(live)
candidate_stat = require_regular(candidate)
if live_stat.st_nlink != 1 or candidate_stat.st_nlink != 1:
raise ContractError('live and candidate must each have one hard link')
if (live_stat.st_dev, live_stat.st_ino) == (
candidate_stat.st_dev,
candidate_stat.st_ino,
):
raise ContractError('live and candidate refer to the same file')
live_metadata = (
live_stat.st_uid,
live_stat.st_gid,
stat.S_IMODE(live_stat.st_mode),
)
candidate_metadata = (
candidate_stat.st_uid,
candidate_stat.st_gid,
stat.S_IMODE(candidate_stat.st_mode),
)
if live_metadata != candidate_metadata:
raise ContractError('owner, group, or permission mode differs')
if live_stat.st_uid != os.geteuid():
raise ContractError('run as the database file owner')
reject_sidecars(live, candidate)
require_rollback_format(live)
require_rollback_format(candidate)
if sha256_file(live) != args.expected_live_sha256:
raise ContractError('live SHA-256 does not match')
if sha256_file(candidate) != args.expected_candidate_sha256:
raise ContractError('candidate SHA-256 does not match')
candidate_info = inspect_database(
candidate,
args.expected_user_version,
)
reject_sidecars(live, candidate)
require_rollback_format(candidate)
if sha256_file(candidate) != args.expected_candidate_sha256:
raise ContractError('candidate changed during validation')
copy_exclusive(live, backup)
if (
sha256_file(backup) != args.expected_live_sha256
or sha256_file(live) != args.expected_live_sha256
):
raise ContractError('live database changed while backup was created')
reject_sidecars(live, candidate)
require_rollback_format(live)
require_rollback_format(candidate)
if sha256_file(live) != args.expected_live_sha256:
raise ContractError('live database changed before replacement')
if sha256_file(candidate) != args.expected_candidate_sha256:
raise ContractError('candidate changed before replacement')
switched = False
try:
os.replace(candidate, live)
switched = True
if sha256_file(live) != args.expected_candidate_sha256:
raise RuntimeError('deployed SHA-256 does not match')
reject_sidecars(live)
require_rollback_format(live)
deployed_info = inspect_database(
live,
args.expected_user_version,
)
reject_sidecars(live)
if deployed_info != candidate_info:
raise RuntimeError('database metadata changed after replacement')
result: dict[str, object] = {
'status': 'passed',
'live': live.name,
'source_candidate': candidate.name,
'backup': backup.name,
'previous_sha256': args.expected_live_sha256,
'deployed_sha256': args.expected_candidate_sha256,
'user_version': deployed_info['user_version'],
'journal_mode': deployed_info['journal_mode'],
'sqlite_runtime': sqlite3.sqlite_version,
'completed_at_utc': datetime.now(timezone.utc)
.isoformat(timespec='seconds')
.replace('+00:00', 'Z'),
}
write_report(report, result)
return result
except BaseException as exc:
if switched:
raise PostCutoverError(
'replacement occurred, but post-check or report failed: '
f'{type(exc).__name__}: {exc}'
) from exc
raise
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser()
parser.add_argument('--live', required=True)
parser.add_argument('--candidate', required=True)
parser.add_argument('--expected-live-sha256', required=True)
parser.add_argument('--expected-candidate-sha256', required=True)
parser.add_argument('--expected-user-version', required=True, type=int)
parser.add_argument('--backup', required=True)
parser.add_argument('--report', required=True)
return parser
def emit(payload: dict[str, object], stream) -> None:
text = json.dumps(
payload,
ensure_ascii=False,
separators=(',', ':'),
) + '\n'
stream.write(text)
stream.flush()
def silence_failed_stream(stream) -> None:
try:
descriptor = stream.fileno()
null_descriptor = os.open(os.devnull, os.O_WRONLY)
except (AttributeError, OSError, ValueError):
return
try:
os.dup2(null_descriptor, descriptor)
finally:
os.close(null_descriptor)
try:
stream.flush()
except (OSError, ValueError):
pass
def emit_best_effort(payload: dict[str, object], stream) -> None:
try:
emit(payload, stream)
except (OSError, ValueError):
silence_failed_stream(stream)
def main() -> int:
try:
args = build_parser().parse_args()
hashes = (
args.expected_live_sha256,
args.expected_candidate_sha256,
)
if any(
len(value) != 64
or any(
character not in '0123456789abcdef'
for character in value
)
for value in hashes
):
raise ContractError(
'SHA-256 must be 64 lowercase hexadecimal characters'
)
result = run(args)
try:
emit(result, sys.stdout)
except (OSError, ValueError) as exc:
silence_failed_stream(sys.stdout)
emit_best_effort(
{
'status': 'attention_required',
'live_may_have_changed': True,
'message': 'cutover report was written, but stdout failed: '
f'{type(exc).__name__}: {exc}',
},
sys.stderr,
)
return 4
return 0
except ContractError as exc:
emit_best_effort(
{'status': 'rejected', 'message': str(exc)},
sys.stderr,
)
return 2
except PostCutoverError as exc:
emit_best_effort(
{
'status': 'attention_required',
'live_may_have_changed': True,
'message': str(exc),
},
sys.stderr,
)
return 4
except KeyboardInterrupt:
emit_best_effort(
{'status': 'interrupted', 'state_unknown': True},
sys.stderr,
)
return 130
except (OSError, sqlite3.Error) as exc:
emit_best_effort(
{'status': 'error', 'message': str(exc)},
sys.stderr,
)
return 3
except Exception as exc:
emit_best_effort(
{'status': 'error', 'message': str(exc)},
sys.stderr,
)
return 3
if __name__ == '__main__':
raise SystemExit(main())
標準出力の失敗も終了コードへ反映する
stream.write()が成功しても、バッファの内容を実際に書き出すflush()で失敗することがあります。そのため、成功JSONのwrite()とflush()を同じtry内で実行します。
さらに、/dev/fullなどへの出力でflushが失敗した場合は、標準出力のファイル記述子を/dev/nullへ付け替えます。これはPython終了処理が同じバッファを再度flushし、意図した終了コード4を別の終了コードへ変えてしまうことを防ぐためです。レポートは標準出力より先に保存されるため、この場合もファイルのレポートを確認できます。
終了コード
| 終了コード | 意味 | 次の行動 |
|---|---|---|
| 0 | 置き換え後の検査、レポート保存、成功JSONの標準出力とflushが完了 | レポートを読み、別に定めた起動手順へ進む |
| 2 | ハッシュ、形式、副ファイル、権限、SQLite検査などの契約違反 | 原因を確認する。直前検査まで進んだ場合は退避コピーが残ることがある |
| 3 | 置き換え前のファイル入出力またはSQLite実行エラー | 作成途中の退避ファイルを含めて状態を確認する |
| 4 | 置き換え後の再検査、レポート作成、または成功JSONの出力に失敗 | アプリを起動せず、稼働用ファイルのハッシュとレポートを確認する |
| 130 | 置き換え前に中断したか、置き換えの成否を確定できない位置で中断 | 状態を推測せず、稼働用ファイルのハッシュを確認する |
argparseが検出した引数不足や未知のオプションは、通常のヘルプ形式で標準エラー出力へ表示され、終了コード2になります。
動作確認用データベースを作る
本番ファイルではなく、新しい空ディレクトリで試します。次のコードは、バージョン1の現行ファイルと、priority列を追加したバージョン2の候補を作ります。
from contextlib import closing
from pathlib import Path
import os
import sqlite3
targets = [Path('current.sqlite'), Path('candidate.sqlite')]
if any(os.path.lexists(path) for path in targets):
raise SystemExit('current.sqlite or candidate.sqlite already exists')
current_schema = '''
CREATE TABLE categories(code TEXT PRIMARY KEY);
CREATE TABLE review_items(
record_id TEXT PRIMARY KEY,
proposed_category TEXT NOT NULL REFERENCES categories(code),
status TEXT NOT NULL
CHECK(status IN ('pending', 'accepted', 'rejected'))
);
INSERT INTO categories(code) VALUES ('billing'), ('delivery');
INSERT INTO review_items(record_id, proposed_category, status)
VALUES
('R-001', 'billing', 'pending'),
('R-002', 'delivery', 'accepted');
PRAGMA user_version = 1;
'''
candidate_schema = '''
CREATE TABLE categories(code TEXT PRIMARY KEY);
CREATE TABLE review_items(
record_id TEXT PRIMARY KEY,
proposed_category TEXT NOT NULL REFERENCES categories(code),
status TEXT NOT NULL
CHECK(status IN ('pending', 'accepted', 'rejected')),
priority INTEGER NOT NULL DEFAULT 0
CHECK(priority BETWEEN 0 AND 3)
);
CREATE INDEX review_pending_priority
ON review_items(priority DESC, record_id)
WHERE status = 'pending';
INSERT INTO categories(code) VALUES ('billing'), ('delivery');
INSERT INTO review_items(
record_id,
proposed_category,
status,
priority
)
VALUES
('R-001', 'billing', 'pending', 0),
('R-002', 'delivery', 'accepted', 0);
PRAGMA user_version = 2;
'''
for path, schema in zip(targets, (current_schema, candidate_schema)):
with closing(sqlite3.connect(path)) as connection:
connection.execute('PRAGMA foreign_keys = ON')
connection.executescript('BEGIN;\n' + schema + '\nCOMMIT;')
同じ利用者、同じディレクトリ、同じumaskで続けて作れば、通常は所有者、グループ、モードが一致します。不一致の場合は、運用規則に従って権限を設定し、検査と候補ハッシュの記録をやり直します。
ハッシュ不一致で停止することを確認する
LIVE_SHA=$(python3 -c 'from pathlib import Path; import hashlib; print(hashlib.sha256(Path("current.sqlite").read_bytes()).hexdigest())')
CANDIDATE_SHA=$(python3 -c 'from pathlib import Path; import hashlib; print(hashlib.sha256(Path("candidate.sqlite").read_bytes()).hexdigest())')
BAD_SHA=$(python3 -c 'import sys; h=sys.argv[1]; print(("1" if h[0] == "0" else "0") + h[1:])' "$CANDIDATE_SHA")
python3 sqlite_cutover.py \
--live current.sqlite \
--candidate candidate.sqlite \
--expected-live-sha256 "$LIVE_SHA" \
--expected-candidate-sha256 "$BAD_SHA" \
--expected-user-version 2 \
--backup rejected-backup.sqlite \
--report rejected-report.json
echo $?標準エラー出力と終了コードは次のようになります。
{"status":"rejected","message":"candidate SHA-256 does not match"}
2この失敗は退避コピーを作る前に検出されます。現行と候補は残り、指定した退避コピーとレポートは作られません。
正しい期待値で切り替える
アプリケーションとSQLiteを開く管理ツールをすべて停止したことを確認してから実行します。
python3 sqlite_cutover.py \
--live current.sqlite \
--candidate candidate.sqlite \
--expected-live-sha256 "$LIVE_SHA" \
--expected-candidate-sha256 "$CANDIDATE_SHA" \
--expected-user-version 2 \
--backup current.before-v2.sqlite \
--report cutover-v2.json
echo $?成功時は標準出力へレポートと同じJSONを1行で出し、終了コード0になります。次のコードで生成物を確認できます。
from contextlib import closing
import json
import sqlite3
from pathlib import Path
report = json.loads(
Path('cutover-v2.json').read_text(encoding='utf-8')
)
print(report['status'])
print(Path('candidate.sqlite').exists())
print(Path('current.before-v2.sqlite').exists())
uri = Path('current.sqlite').resolve().as_uri() + '?mode=ro'
with closing(sqlite3.connect(uri, uri=True)) as connection:
print(connection.execute('PRAGMA user_version').fetchone()[0])
passed
False
True
2候補の元パスが存在しないのは、そのファイルをcurrent.sqliteという名前へ移したためです。退避コピーのSHA-256は、レポートのprevious_sha256と一致します。
終了コード4または130ではハッシュから状態を確認する
アプリを停止した状態を保ち、稼働用パスのSHA-256を計算します。
python3 - current.sqlite <<'PY'
import hashlib
import sys
digest = hashlib.sha256()
with open(sys.argv[1], 'rb') as source:
while chunk := source.read(1024 * 1024):
digest.update(chunk)
print(digest.hexdigest())
PYCANDIDATE_SHAと一致する場合、候補の内容が稼働用パスにあります。レポートと後段検査の失敗理由を確認します。LIVE_SHAと一致する場合、切り替え前の内容です。- どちらとも一致しない場合は、起動や再実行をせず、メインファイルと副ファイルを保全して調査します。
アプリ起動後に退避コピーを戻すと、新しい更新を失う可能性があります。この退避コピーは、自動ロールバックを許可するものではありません。
適用上の制約
- プロセス間ロックは実装していません。コマンドを並行実行せず、対象ディレクトリをほかの処理から変更しない運用が必要です。
- 単一のSQLiteメインファイルだけを対象とします。
ATTACHしたデータベースや外部ファイルは切り替えません。 - WAL形式は、副ファイルが存在しない場合も対象外です。
- ネットワークファイルシステム、Windows、コンテナ間で共有するボリュームにはそのまま適用しません。
- 退避コピーとレポートには
fsync()を使いますが、ディレクトリエントリ、ストレージ装置、突然の電源断まで含む耐久性は保証しません。 - コードは候補の構造整合性とバージョンを再確認しますが、移行固有の書き込み試験や検索結果の突合は再実行しません。記録済みの候補ハッシュによって、事前検査したファイルと切り替えるファイルを結び付けます。
status=passedが示すのは、指定したデータベースファイルの切り替えと再検査が完了したことです。アプリケーション全体の正常稼働は、別の起動手順とヘルスチェックで確認します。