第171回　SQLiteを旧版へ戻す前に止める：Pythonで作る変更検知付きロールバックゲート

第170回では、検証済みのSQLiteをハッシュで固定し、稼働用ファイルへ原子的に切り替えました。ただし、切り替え後に問題が見つかったからといって、退避した旧ファイルへ直ちに戻してよいとは限りません。

新しいデータベースへ確定済みの書き込みがある状態で旧版を配置すると、その変更を失います。今回はロールバック自体を実行せず、現行ファイルと退避ファイルが切り替え時の状態から変わっていないかを判定するゲートを作ります。

eligible=trueは、指定した検査を実行した時点で条件を満たしたという意味です。ロールバックの安全性を保証したり、検査後の状態を固定したりするものではありません。

ロールバック実行と判定を分ける

確認結果	意味	対応
現行のSHA-256が一致	切り替え直後に記録したバイト列と同じ	ほかの検査へ進む
現行のSHA-256が不一致	切り替え後にファイルが変化したか、別ファイルを指定した	旧版で置き換えず、差分を調査する
退避ファイルのSHA-256が不一致	退避ファイルの変更、破損、取り違えを否定できない	ロールバック候補として使わない
副ファイルが存在	WAL、共有メモリ、ジャーナルが残っている	接続状態と正しい停止手順を確認する
整合性、外部キー、バージョンが不一致	期待したSQLiteではない可能性がある	原因を確認するまで停止する

SHA-256の一致は、切り替え後に一度も処理が動かなかったことまでは証明しません。たとえばSQLite外部のAPI呼び出し、メッセージ送信、キャッシュ更新は検出できません。

今回の実行契約

項目	仕様
動作環境	Python 3.10以上、SQLite 3.22.0以上、ローカルのPOSIXファイルシステム
事前条件	アプリケーションを停止し、現行・退避SQLiteへの接続をすべて閉じている
切り替え記録	切り替え時点で保存した現行・退避ファイルのSHA-256とuser_version
ファイル条件	通常ファイルであり、シンボリックリンク、複数ハードリンク、同じinodeを拒否する
副ファイル	-wal、-shm、-journalが一つでもあれば不合格
SQLite検査	integrity_check(1)、foreign_key_check、user_version
出力	既存ファイルを上書きしないJSONレポート
対象外	ファイルの置換、アプリの再起動、外部システムの副作用、業務データの意味上の正しさ

読み取りにはmode=ro&immutable=1を使います。SQLite公式のURI仕様によると、immutableはロックと変更検知を省略する指定です。稼働中のデータベースへ使ってはいけません。WALモードの読み取り条件はSQLite公式のWAL仕様でも説明されています。

この例のcutover_record.jsonは独立した最小形式です。第170回のレポートをそのまま読み込めるとは限りません。実運用では切り替え前に項目の対応を決め、障害が起きてから期待ハッシュを現行ファイルから作り直さないでください。記録自体を変更できる人がデータベースも変更できる環境では、SHA-256は改ざんの証明になりません。

1．動作確認用の入力を作る

次のコードをmake_demo.pyとして保存します。実データは使わず、新しいdemoディレクトリへ旧版、切り替え後、切り替え記録を作ります。demoが存在する場合は上書きしません。

from __future__ import annotations

import hashlib
import json
import os
import shutil
import sqlite3
from pathlib import Path


ROOT = Path('demo')
PREVIOUS = ROOT / 'previous.sqlite'
ACTIVE = ROOT / 'active.sqlite'
RECORD = ROOT / 'cutover_record.json'


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open('rb') as source_file:
        for chunk in iter(lambda: source_file.read(1024 * 1024), b''):
            digest.update(chunk)
    return digest.hexdigest()


if os.path.lexists(ROOT):
    raise SystemExit('demo already exists; use a new working directory')
ROOT.mkdir(mode=0o700)

connection = sqlite3.connect(PREVIOUS)
try:
    connection.executescript('''
        CREATE TABLE review_items (
            id INTEGER PRIMARY KEY,
            status TEXT NOT NULL
                CHECK (status IN ('pending', 'accepted'))
        );

        INSERT INTO review_items(id, status)
        VALUES (1, 'pending'), (2, 'accepted');

        PRAGMA user_version = 1;
    ''')
    connection.commit()
finally:
    connection.close()

shutil.copyfile(PREVIOUS, ACTIVE)
connection = sqlite3.connect(ACTIVE)
try:
    connection.executescript('''
        ALTER TABLE review_items
            ADD COLUMN priority INTEGER NOT NULL DEFAULT 0;

        CREATE INDEX review_items_priority_idx
            ON review_items(priority, id);

        PRAGMA user_version = 2;
    ''')
    connection.commit()
finally:
    connection.close()

record = {
    'format': 'sqlite-cutover-record-v1',
    'active': {
        'sha256': sha256_file(ACTIVE),
        'user_version': 2,
    },
    'previous': {
        'sha256': sha256_file(PREVIOUS),
        'user_version': 1,
    },
}

with RECORD.open('x', encoding='utf-8') as destination:
    json.dump(record, destination, ensure_ascii=False, indent=2, sort_keys=True)
    destination.write(chr(10))
    destination.flush()
    os.fsync(destination.fileno())

print(json.dumps(
    {
        'active': str(ACTIVE),
        'previous': str(PREVIOUS),
        'record': str(RECORD),
        'status': 'created',
    },
    ensure_ascii=False,
    separators=(',', ':'),
    sort_keys=True,
))

実行します。

python3 make_demo.py

標準出力は次の1行です。SHA-256はファイルへ記録されるため、固定値を例示しません。

{"active":"demo/active.sqlite","previous":"demo/previous.sqlite","record":"demo/cutover_record.json","status":"created"}

2．ロールバックゲートを実装する

次のコードをrollback_ready.pyとして保存します。検査は段階的に進み、ハッシュ不一致や副ファイルを見つけた場合は、想定外のSQLiteを開かずに不合格レポートを確定します。

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sqlite3
import stat
import sys
from pathlib import Path
from typing import Any


MAX_RECORD_BYTES = 16 * 1024
MIN_SQLITE = (3, 22, 0)
SIDECAR_SUFFIXES = ('-wal', '-shm', '-journal')


class InputError(Exception):
    pass


class OperationalError(Exception):
    pass


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description='Inspect whether an unchanged SQLite cutover can be rolled back.'
    )
    parser.add_argument('--active', required=True, type=Path)
    parser.add_argument('--previous', required=True, type=Path)
    parser.add_argument('--record', required=True, type=Path)
    parser.add_argument('--report', required=True, type=Path)
    return parser.parse_args()


def lstat_regular(path: Path, *, single_link: bool) -> os.stat_result:
    try:
        file_stat = os.lstat(path)
    except FileNotFoundError as exc:
        raise InputError(f'file not found: {path}') from exc
    except OSError as exc:
        raise OperationalError(f'cannot inspect {path}: {exc}') from exc

    if not stat.S_ISREG(file_stat.st_mode):
        raise InputError(f'not a regular file: {path}')
    if single_link and file_stat.st_nlink != 1:
        raise InputError(f'database must have exactly one hard link: {path}')
    return file_stat


def signature(file_stat: os.stat_result) -> tuple[int, int, int, int, int]:
    return (
        file_stat.st_dev,
        file_stat.st_ino,
        file_stat.st_size,
        file_stat.st_mtime_ns,
        file_stat.st_ctime_ns,
    )


def reject_constant(token: str) -> None:
    raise InputError(f'non-standard JSON number: {token}')


def object_without_duplicates(
    pairs: list[tuple[str, Any]],
) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise InputError(f'duplicate JSON key: {key}')
        result[key] = value
    return result


def valid_sha256(value: Any) -> bool:
    return (
        type(value) is str
        and len(value) == 64
        and all(character in '0123456789abcdef' for character in value)
    )


def validate_record(value: Any) -> dict[str, Any]:
    if type(value) is not dict:
        raise InputError('record must be a JSON object')
    if set(value) != {'format', 'active', 'previous'}:
        raise InputError('record has missing or unknown top-level keys')
    if value['format'] != 'sqlite-cutover-record-v1':
        raise InputError('unsupported record format')

    for role in ('active', 'previous'):
        section = value[role]
        if type(section) is not dict or set(section) != {
            'sha256', 'user_version'
        }:
            raise InputError(f'invalid {role} section')
        if not valid_sha256(section['sha256']):
            raise InputError(f'invalid {role} SHA-256')
        version = section['user_version']
        if type(version) is not int or not 0 <= version <= 2_147_483_647:
            raise InputError(f'invalid {role} user_version')
    return value


def read_record(path: Path) -> dict[str, Any]:
    file_stat = lstat_regular(path, single_link=False)
    if file_stat.st_size > MAX_RECORD_BYTES:
        raise InputError('record is too large')
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise OperationalError(f'cannot read record: {exc}') from exc
    if len(raw) > MAX_RECORD_BYTES:
        raise InputError('record is too large')
    try:
        text = raw.decode('utf-8')
    except UnicodeDecodeError as exc:
        raise InputError('record is not UTF-8') from exc
    try:
        value = json.loads(
            text,
            object_pairs_hook=object_without_duplicates,
            parse_constant=reject_constant,
        )
    except json.JSONDecodeError as exc:
        message = (
            f'invalid record JSON at line {exc.lineno}, column {exc.colno}'
        )
        raise InputError(message) from exc
    return validate_record(value)


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    try:
        with path.open('rb') as source_file:
            for chunk in iter(
                lambda: source_file.read(1024 * 1024), b''
            ):
                digest.update(chunk)
    except OSError as exc:
        raise OperationalError(f'cannot hash {path}: {exc}') from exc
    return digest.hexdigest()


def sqlite_check(path: Path) -> dict[str, Any]:
    try:
        uri = path.resolve(strict=True).as_uri() + '?mode=ro&immutable=1'
    except OSError as exc:
        raise OperationalError(f'cannot resolve {path}: {exc}') from exc

    result: dict[str, Any] = {
        'integrity_ok': False,
        'foreign_keys_ok': False,
        'user_version': None,
        'error': None,
    }
    connection: sqlite3.Connection | None = None
    try:
        connection = sqlite3.connect(uri, uri=True, timeout=1.0)
        connection.execute('PRAGMA query_only = ON')
        integrity_rows = connection.execute(
            'PRAGMA integrity_check(1)'
        ).fetchall()
        result['integrity_ok'] = integrity_rows == [('ok',)]
        result['foreign_keys_ok'] = (
            connection.execute('PRAGMA foreign_key_check').fetchone() is None
        )
        version_row = connection.execute('PRAGMA user_version').fetchone()
        result['user_version'] = (
            None if version_row is None else version_row[0]
        )
    except sqlite3.Error as exc:
        result['error'] = type(exc).__name__
    finally:
        if connection is not None:
            try:
                connection.close()
            except sqlite3.Error as exc:
                if result['error'] is None:
                    result['error'] = type(exc).__name__
    return result


def find_sidecars(active: Path, previous: Path) -> list[str]:
    found: list[str] = []
    for role, path in (('active', active), ('previous', previous)):
        for suffix in SIDECAR_SUFFIXES:
            if os.path.lexists(os.fspath(path) + suffix):
                found.append(f'{role}:{suffix}')
    return found


def add_check(
    report: dict[str, Any], name: str, passed: bool
) -> None:
    report['checks'][name] = passed
    if not passed:
        report['failures'].append(name)


def finish(report: dict[str, Any]) -> dict[str, Any]:
    report['eligible'] = not report['failures']
    report['status'] = 'passed' if report['eligible'] else 'failed'
    return report


def inspect(
    active: Path,
    previous: Path,
    record: dict[str, Any],
) -> dict[str, Any]:
    active_before = lstat_regular(active, single_link=True)
    previous_before = lstat_regular(previous, single_link=True)
    if (active_before.st_dev, active_before.st_ino) == (
        previous_before.st_dev,
        previous_before.st_ino,
    ):
        raise InputError('active and previous refer to the same file')

    report: dict[str, Any] = {
        'format': 'sqlite-rollback-readiness-v1',
        'status': 'failed',
        'eligible': False,
        'checks': {},
        'failures': [],
        'observed': {},
    }

    sidecars_before = find_sidecars(active, previous)
    report['observed']['sidecars_before'] = sidecars_before
    add_check(report, 'no_sidecars_before', not sidecars_before)
    if report['failures']:
        return finish(report)

    active_hash = sha256_file(active)
    previous_hash = sha256_file(previous)
    active_after_hash = lstat_regular(active, single_link=True)
    previous_after_hash = lstat_regular(previous, single_link=True)

    report['observed']['active'] = {
        'sha256': active_hash,
        'size': active_after_hash.st_size,
    }
    report['observed']['previous'] = {
        'sha256': previous_hash,
        'size': previous_after_hash.st_size,
    }
    add_check(
        report,
        'active_stable_during_initial_hash',
        signature(active_before) == signature(active_after_hash),
    )
    add_check(
        report,
        'previous_stable_during_initial_hash',
        signature(previous_before) == signature(previous_after_hash),
    )
    add_check(
        report,
        'active_hash_matches_record',
        active_hash == record['active']['sha256'],
    )
    add_check(
        report,
        'previous_hash_matches_record',
        previous_hash == record['previous']['sha256'],
    )
    if report['failures']:
        return finish(report)

    active_sqlite = sqlite_check(active)
    previous_sqlite = sqlite_check(previous)
    report['observed']['active']['sqlite'] = active_sqlite
    report['observed']['previous']['sqlite'] = previous_sqlite

    add_check(
        report, 'active_sqlite_readable', active_sqlite['error'] is None
    )
    add_check(
        report, 'previous_sqlite_readable', previous_sqlite['error'] is None
    )
    add_check(report, 'active_integrity_ok', active_sqlite['integrity_ok'])
    add_check(report, 'previous_integrity_ok', previous_sqlite['integrity_ok'])
    add_check(
        report, 'active_foreign_keys_ok', active_sqlite['foreign_keys_ok']
    )
    add_check(
        report, 'previous_foreign_keys_ok', previous_sqlite['foreign_keys_ok']
    )
    add_check(
        report,
        'active_user_version_matches_record',
        active_sqlite['user_version'] == record['active']['user_version'],
    )
    add_check(
        report,
        'previous_user_version_matches_record',
        previous_sqlite['user_version'] == record['previous']['user_version'],
    )

    active_hash_after = sha256_file(active)
    previous_hash_after = sha256_file(previous)
    active_after = lstat_regular(active, single_link=True)
    previous_after = lstat_regular(previous, single_link=True)
    sidecars_after = find_sidecars(active, previous)
    report['observed']['sidecars_after'] = sidecars_after

    add_check(
        report,
        'active_unchanged_during_inspection',
        active_hash_after == active_hash
        and signature(active_after) == signature(active_before),
    )
    add_check(
        report,
        'previous_unchanged_during_inspection',
        previous_hash_after == previous_hash
        and signature(previous_after) == signature(previous_before),
    )
    add_check(report, 'no_sidecars_after', not sidecars_after)
    return finish(report)


def remove_partial_report(path: Path, created: bool) -> None:
    if created:
        try:
            path.unlink()
        except OSError:
            pass


def write_new_json(path: Path, value: dict[str, Any]) -> None:
    serialized = json.dumps(
        value,
        ensure_ascii=False,
        indent=2,
        sort_keys=True,
    ) + chr(10)
    created = False
    try:
        with path.open('x', encoding='utf-8') as destination:
            created = True
            destination.write(serialized)
            destination.flush()
            os.fsync(destination.fileno())
    except FileExistsError as exc:
        raise InputError(f'report already exists: {path}') from exc
    except OSError as exc:
        remove_partial_report(path, created)
        raise OperationalError(f'cannot write report: {exc}') from exc
    except BaseException:
        remove_partial_report(path, created)
        raise


def emit(stream: Any, value: dict[str, Any]) -> None:
    print(
        json.dumps(
            value,
            ensure_ascii=False,
            separators=(',', ':'),
            sort_keys=True,
        ),
        file=stream,
    )


def main() -> int:
    args = parse_args()
    try:
        if sqlite3.sqlite_version_info < MIN_SQLITE:
            raise OperationalError('SQLite 3.22.0 or newer is required')
        if os.path.lexists(args.report):
            raise InputError(f'report already exists: {args.report}')
        record = read_record(args.record)
        report = inspect(args.active, args.previous, record)
        write_new_json(args.report, report)
    except InputError as exc:
        emit(sys.stderr, {'error': str(exc), 'status': 'input_error'})
        return 2
    except OperationalError as exc:
        emit(sys.stderr, {'error': str(exc), 'status': 'operational_error'})
        return 3
    except KeyboardInterrupt:
        emit(
            sys.stderr,
            {'error': 'interrupted', 'status': 'operational_error'},
        )
        return 130
    except Exception as exc:
        emit(
            sys.stderr,
            {'error': type(exc).__name__, 'status': 'unexpected_error'},
        )
        return 3

    emit(
        sys.stdout,
        {
            'eligible': report['eligible'],
            'report': os.fspath(args.report),
            'status': report['status'],
        },
    )
    return 0 if report['eligible'] else 1


if __name__ == '__main__':
    raise SystemExit(main())

3．変更のない状態を判定する

アプリケーションを停止した状態を想定し、次の順序で実行します。

python3 rollback_ready.py --active demo/active.sqlite --previous demo/previous.sqlite --record demo/cutover_record.json --report demo/report-ok.json
echo $?

標準出力と終了コードは次のとおりです。

{"eligible":true,"report":"demo/report-ok.json","status":"passed"}
0

レポートの主要項目だけを確認します。

python3 - <<'PY'
import json
from pathlib import Path

report = json.loads(
    Path('demo/report-ok.json').read_text(encoding='utf-8')
)
print(json.dumps(
    {
        'eligible': report['eligible'],
        'failures': report['failures'],
        'status': report['status'],
    },
    ensure_ascii=False,
    sort_keys=True,
))
PY

{"eligible": true, "failures": [], "status": "passed"}

この状態でもアプリケーションを再起動してはいけません。判定後に書き込みが始まると、レポートは直ちに古くなります。

4．切り替え後の変更を再現する

現行ファイルの1行を更新し、確定します。これは動作確認用データに対してだけ実行します。

python3 - <<'PY'
import sqlite3

connection = sqlite3.connect('demo/active.sqlite')
try:
    cursor = connection.execute(
        'UPDATE review_items SET priority = 2 WHERE id = 1'
    )
    if cursor.rowcount != 1:
        raise RuntimeError('expected exactly one updated row')
    connection.commit()
finally:
    connection.close()
PY

既存レポートは上書きできないため、別の出力先を指定します。

python3 rollback_ready.py --active demo/active.sqlite --previous demo/previous.sqlite --record demo/cutover_record.json --report demo/report-changed.json

標準出力は次の1行で、終了コードは1です。

{"eligible":false,"report":"demo/report-changed.json","status":"failed"}

失敗理由を確認します。

python3 - <<'PY'
import json
from pathlib import Path

report = json.loads(
    Path('demo/report-changed.json').read_text(encoding='utf-8')
)
print(report['failures'])
PY

['active_hash_matches_record']

ハッシュが一致しないため、スクリプトは現行SQLiteを開かずに停止します。更新内容が正しいかどうかに関係なく、旧版で置き換えても変更を失わないという前提が崩れたためです。

終了コードを後続処理の条件にする

終了コード	意味	対応
0	検査時点で全条件を満たした	アプリを停止したまま、別途承認された手順で判断を続ける
1	検査を完了したが不合格	自動ロールバックしない
2	入力、記録形式、ファイル指定、既存レポートの問題	実行条件を修正して最初から検査する
3	SQLiteのバージョン、ファイル入出力、予期しない実行エラー	合格として扱わず、環境を確認する
130	Ctrl+Cによる中断	生成途中のレポートを使わず再実行する

引数の不足や誤記はargparseの使用方法を標準エラーへ表示し、終了コード2になります。それ以外のスクリプト内エラーは1行のJSONとして標準エラーへ出します。

このゲートで判断できないこと

• SHA-256の不一致から、変更した処理や変更内容までは特定できない
• 切り替え記録も書き換えられた場合、ハッシュ一致だけでは改ざんを検出できない
• 外部API、メール、ジョブキュー、添付データベース、別ファイルへの副作用は検査しない
• integrity_checkとforeign_key_checkは、業務上の値の正しさを保証しない
• 検査完了から実際のロールバックまでの間に状態が変わることを防止しない
• 親ディレクトリの差し替えなど、悪意ある同時ファイル操作への防御を目的としない

現行ハッシュが変わっていた場合は、旧版への置換ではなく、現行データベースの隔離コピーを作り、切り替え後の変更を特定して引き継ぐ方法を検討します。終了コード0の場合も、アプリケーションを停止したまま実際の置換直前に再検査し、承認済みのロールバック手順と組み合わせる必要があります。