第176回 CLIへ渡す環境変数を絞る:os.execveで作る許可リスト実行ラッパー

第172回から第175回では、複数CLIの順次実行、二重起動の防止、再試行、タイムアウト時の停止を扱いました。実行方法を整えても、子へ渡す環境変数が未整理では、不要な設定や秘密値まで引き継ぐ可能性があります。

今回は、親から引き継ぐ変数と固定値をJSONで指定し、それ以外を渡さずにCLIを起動します。対象はPython 3.10以上を実行するPOSIX環境です。外部ライブラリは使いません。

この仕組みが制限するのは、実行対象へ渡す環境変数です。秘密管理、利用者認証、ファイルやネットワークの隔離までは行いません。

現在のプロセスを実行対象へ置き換える

os.execve()は、現在のプロセスを指定したプログラムへ置き換えます。渡した環境マッピングは現在の環境への追加ではなく、新しいプログラムの環境として使われます。execve()自身はPATHから実行ファイルを検索しません。

実行に成功するとPythonラッパーへは戻りません。POSIX環境ではPIDとカレントディレクトリが維持されます。開いているファイル記述子も、close-on-execが設定されたものを除いて引き継がれます。このラッパーは標準入力、標準出力、標準エラー出力を付け替えません。

方式 向いている用途 今回の扱い
subprocess.Popen(..., env=...) 親が子の完了を待ち、その後に記録や後処理を行う 使わない
os.execve(path, argv, env) 環境を確定した後、その場で実行対象へ置き換わる 採用する

execve()後に監査レポートを書く場合は、ラッパーの外側に監督プロセスが必要です。第175回のタイムアウトラッパーなどから今回のラッパーを起動すれば、置換後も監督対象のPIDは変わりません。

今回の実行契約

項目 仕様
ポリシー BOMなしUTF-8のJSON。最大65,536バイトで、最上位項目はinherit、required、setの三つだけ
inherit 親に存在する場合だけ値を引き継ぐ変数名の配列
required 親に存在しなければ実行を止める変数名の配列。すべてinheritにも含める
set ポリシーで固定する文字列値。inheritとの同名指定は拒否する
変数名 英字またはアンダースコアで始まり、以後は英数字またはアンダースコア。重複は不可
上限 子へ渡す変数は最大64件、値は1件8,192バイト。環境全体は各名前=値と終端を数えて65,536バイト
実行ファイル --の後ろに絶対パスで指定する。シェルとPATH検索は使わない
入出力 標準入力、標準出力、標準エラー出力を付け替えない
成功時 ラッパーを対象CLIへ置き換える。独自の成功メッセージは出さない
失敗時 環境変数の値を含まない1行JSONを標準エラー出力へ出す

65,536バイトは、このサンプルで決めた上限です。OSには引数と環境を合わせた別の上限があります。長い引数列などのために起動を拒否された場合は終了コード126にします。

1.環境ポリシーを作る

次をenv-policy.jsonとして保存します。

{
  "inherit": ["JOB_BATCH_ID"],
  "required": ["JOB_BATCH_ID"],
  "set": {
    "APP_MODE": "batch",
    "PYTHONUTF8": "1"
  }
}

JOB_BATCH_IDは実行元から受け取り、欠けていれば止めます。APP_MODEとPYTHONUTF8は固定値です。

ポリシーファイルへ秘密値を直接書かないでください。対象CLIが秘密値を必要とする場合は、その変数名をinheritとrequiredへ追加し、別の方法で実行元へ設定します。

PATH、HOME、ロケール、プロキシー関連変数は自動では渡りません。必要性と値を確認してから追加します。LD_PRELOAD、LD_LIBRARY_PATH、PYTHONPATHなどは読み込み動作を変えるため、普段から設定されているという理由だけで許可しません。

2.確認用CLIを用意する

次をinspect_env.pyとして保存します。値を表示するのは動作確認用のバッチIDだけです。実運用で秘密値を表示してはいけません。

import json
import os

result = {
    "job_batch_id": os.environ.get("JOB_BATCH_ID"),
    "app_mode": os.environ.get("APP_MODE"),
    "api_token_present": "API_TOKEN" in os.environ,
    "path_present": "PATH" in os.environ,
}
print(json.dumps(result, ensure_ascii=False, separators=(",", ":")))

未許可のAPI_TOKENとPATHについては、値ではなく存在の有無だけを返します。

3.許可リスト実行ラッパーを作る

次をenv_exec.pyとして保存します。

from __future__ import annotations

import json
import os
import re
import sys
from pathlib import Path
from typing import Any

MAX_POLICY_BYTES = 65_536
MAX_VARIABLES = 64
MAX_VALUE_BYTES = 8_192
MAX_ENV_BYTES = 65_536
NAME_PATTERN = re.compile(r"[A-Za-z_][A-Za-z0-9_]*\Z")
POLICY_KEYS = {"inherit", "required", "set"}
USAGE = "usage: env_exec.py POLICY.json -- /absolute/executable [arg ...]"


class ContractError(Exception):
    pass


def reject_number(_value: str) -> None:
    raise ContractError("このポリシーでJSON数値は使用できません")


def reject_constant(value: str) -> None:
    raise ContractError(f"JSON定数 {value} は使用できません")


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ContractError(f"JSONキーが重複しています: {key!r}")
        result[key] = value
    return result


def validate_name(name: Any, label: str) -> str:
    if type(name) is not str or NAME_PATTERN.fullmatch(name) is None:
        raise ContractError(
            f"{label}の変数名は英字または_で始まる英数字と_だけにしてください"
        )
    return name


def validate_name_list(value: Any, label: str) -> list[str]:
    if type(value) is not list:
        raise ContractError(f"{label}は配列にしてください")
    if len(value) > MAX_VARIABLES:
        raise ContractError(f"{label}は{MAX_VARIABLES}件以下にしてください")

    result: list[str] = []
    seen: set[str] = set()
    for item in value:
        name = validate_name(item, label)
        if name in seen:
            raise ContractError(f"{label}内で変数名が重複しています: {name}")
        seen.add(name)
        result.append(name)
    return result


def encoded_size(value: str, label: str) -> int:
    if "\0" in value:
        raise ContractError(f"{label}にNUL文字は使用できません")
    try:
        return len(os.fsencode(value))
    except UnicodeEncodeError as exc:
        raise ContractError(f"{label}をOS向けに符号化できません") from exc


def validate_value(name: str, value: Any) -> str:
    if type(value) is not str:
        raise ContractError(f"set.{name}は文字列にしてください")
    size = encoded_size(value, name)
    if size > MAX_VALUE_BYTES:
        raise ContractError(
            f"環境変数{name}の値は{MAX_VALUE_BYTES}バイト以下にしてください"
        )
    return value


def load_policy(path: Path) -> tuple[list[str], list[str], dict[str, str]]:
    try:
        with path.open("rb") as handle:
            raw = handle.read(MAX_POLICY_BYTES + 1)
    except OSError as exc:
        detail = exc.strerror or type(exc).__name__
        raise ContractError(f"ポリシーを読み取れません: {detail}") from exc

    if len(raw) > MAX_POLICY_BYTES:
        raise ContractError(
            f"ポリシーは{MAX_POLICY_BYTES}バイト以下にしてください"
        )
    if raw.startswith(b"\xef\xbb\xbf"):
        raise ContractError("ポリシーはBOMなしUTF-8にしてください")

    try:
        text = raw.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise ContractError("ポリシーはBOMなしUTF-8にしてください") from exc

    try:
        data = json.loads(
            text,
            object_pairs_hook=unique_object,
            parse_int=reject_number,
            parse_float=reject_number,
            parse_constant=reject_constant,
        )
    except json.JSONDecodeError as exc:
        raise ContractError(
            f"JSON構文エラーです: line={exc.lineno}, column={exc.colno}"
        ) from exc
    except RecursionError as exc:
        raise ContractError("JSONの入れ子が深すぎます") from exc

    if type(data) is not dict:
        raise ContractError("ポリシーの最上位はオブジェクトにしてください")

    actual_keys = set(data)
    if actual_keys != POLICY_KEYS:
        missing = sorted(POLICY_KEYS - actual_keys)
        unknown = sorted(actual_keys - POLICY_KEYS)
        raise ContractError(
            f"ポリシー項目が一致しません: missing={missing}, unknown={unknown}"
        )

    inherit = validate_name_list(data["inherit"], "inherit")
    required = validate_name_list(data["required"], "required")

    fixed_raw = data["set"]
    if type(fixed_raw) is not dict:
        raise ContractError("setはオブジェクトにしてください")
    if len(fixed_raw) > MAX_VARIABLES:
        raise ContractError(f"setは{MAX_VARIABLES}件以下にしてください")

    fixed: dict[str, str] = {}
    for raw_name, raw_value in fixed_raw.items():
        name = validate_name(raw_name, "set")
        fixed[name] = validate_value(name, raw_value)

    inherit_set = set(inherit)
    required_set = set(required)
    fixed_set = set(fixed)

    if not required_set <= inherit_set:
        names = sorted(required_set - inherit_set)
        raise ContractError(f"requiredはinheritにも含めてください: {names}")
    if inherit_set & fixed_set:
        names = sorted(inherit_set & fixed_set)
        raise ContractError(f"inheritとsetで変数名が重複しています: {names}")
    if len(inherit_set | fixed_set) > MAX_VARIABLES:
        raise ContractError(
            f"子へ渡す環境変数は合計{MAX_VARIABLES}件以下にしてください"
        )

    return inherit, required, fixed


def build_environment(
    inherit: list[str], required: list[str], fixed: dict[str, str]
) -> dict[str, str]:
    missing = [name for name in required if name not in os.environ]
    if missing:
        raise ContractError(f"必須の環境変数がありません: {missing}")

    child_env: dict[str, str] = {}
    for name in inherit:
        if name in os.environ:
            value = os.environ[name]
            validate_value(name, value)
            child_env[name] = value

    child_env.update(fixed)
    total = sum(
        encoded_size(name, name) + 1 + encoded_size(value, name) + 1
        for name, value in child_env.items()
    )
    if total > MAX_ENV_BYTES:
        raise ContractError(
            f"子へ渡す環境変数は合計{MAX_ENV_BYTES}バイト以下にしてください"
        )
    return child_env


def parse_cli(argv: list[str]) -> tuple[Path, list[str]] | None:
    if argv in (["-h"], ["--help"]):
        return None
    if len(argv) < 3 or argv[1] != "--":
        raise ContractError(USAGE)

    command = argv[2:]
    executable = command[0]
    if not os.path.isabs(executable):
        raise ContractError("実行ファイルは絶対パスで指定してください")
    return Path(argv[0]), command


def emit_error(kind: str, message: str, **extra: Any) -> None:
    payload: dict[str, Any] = {
        "status": "error",
        "kind": kind,
        "message": message,
    }
    payload.update(extra)
    raw = (
        json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n"
    ).encode("utf-8", errors="backslashreplace")
    sys.stderr.buffer.write(raw)
    sys.stderr.buffer.flush()


def main(argv: list[str]) -> int:
    try:
        parsed = parse_cli(argv)
        if parsed is None:
            print(USAGE)
            return 0

        policy_path, command = parsed
        inherit, required, fixed = load_policy(policy_path)
        child_env = build_environment(inherit, required, fixed)

        try:
            os.execve(command[0], command, child_env)
        except FileNotFoundError:
            emit_error(
                "exec_not_found",
                "実行ファイルまたは指定されたインタープリターが見つかりません",
                executable=command[0],
            )
            return 127
        except PermissionError:
            emit_error(
                "exec_not_permitted",
                "実行ファイルを起動する権限がありません",
                executable=command[0],
            )
            return 126
        except OSError as exc:
            emit_error(
                "exec_failed",
                "実行ファイルを起動できません",
                executable=command[0],
                errno=exc.errno,
            )
            return 126
    except ContractError as exc:
        emit_error("contract_error", str(exc))
        return 2
    except KeyboardInterrupt:
        emit_error("interrupted", "実行前に中断されました")
        return 130


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))

object_pairs_hook、parse_int、parse_float、parse_constantを使っています。通常は後の値が残る重複キーを拒否し、このポリシーで使わないJSON数値と、標準JSON外のNaNやInfinityも受け付けません。

ポリシーは上限より1バイトだけ多く読みます。上限超過を判定するために、ファイル全体を読み込む必要はありません。

os.fsencode()で数えるのは、POSIX上でPythonが環境変数をOSへ渡す際の符号化に合わせるためです。ただし、この検査はOSによる起動成功を保証しません。

4.許可した変数だけが見えることを確認する

現在使っているPythonインタープリターのパスを取得します。sys.executableを取得できない場合や、絶対パスでない場合は止めます。

PYTHON_BIN="$(python3 -c 'import sys; print(sys.executable or "")' )"
case "$PYTHON_BIN" in
  /*) ;;
  *) exit 1 ;;
esac

動作確認用の値を親へ設定して実行します。API_TOKENの値は例示用であり、実在する認証情報ではありません。

JOB_BATCH_ID=batch-20261023 API_TOKEN=sample-not-passed \
  "$PYTHON_BIN" env_exec.py env-policy.json -- \
  "$PYTHON_BIN" inspect_env.py

標準出力は次の1行です。

{"job_batch_id":"batch-20261023","app_mode":"batch","api_token_present":false,"path_present":false}

JOB_BATCH_IDと固定値は渡り、未許可のAPI_TOKENとPATHは渡っていません。終了コードは確認用CLIの0です。

execve()が渡すのは許可した環境ですが、起動後のプログラムが環境変数を追加または変更することはあります。対象CLI内の環境が将来も同じ件数である、といった判定には使いません。

5.必須変数の不足と対象CLIの失敗を確認する

必須のJOB_BATCH_IDを外したサブシェルで実行します。

(
  unset JOB_BATCH_ID
  "$PYTHON_BIN" env_exec.py env-policy.json -- \
    "$PYTHON_BIN" inspect_env.py
)
echo $?

標準エラー出力と終了コードは次のとおりです。環境変数の値は出力しません。

{"status":"error","kind":"contract_error","message":"必須の環境変数がありません: ['JOB_BATCH_ID']"}
2

実行対象が終了コード7を返す場合は、その値が呼び出し元へ返ります。

JOB_BATCH_ID=batch-20261023 \
  "$PYTHON_BIN" env_exec.py env-policy.json -- \
  "$PYTHON_BIN" -c 'import sys; sys.exit(7)'
echo $?
7
終了コード 意味
対象CLIの終了コード execve()成功後は対象CLIが直接返す。0に限定しない
2 引数、ポリシー、必須環境変数の契約違反。ただし対象CLI自身も2を返し得る
126 権限不足、形式不正、引数と環境の過大などにより対象を起動できなかった
127 実行ファイル、またはスクリプトのshebangが指すインタープリターが見つからなかった
130 execve()より前にラッパーがCtrl+Cで中断された

execve()後のCtrl+CやSIGTERMは、置き換わった対象プログラムが直接受けます。最終的な状態は、そのプログラムのシグナル処理に従います。

ラッパー由来の終了コード2と対象CLI由来の2は、終了コードだけでは区別できません。必要なら、標準エラー出力のJSONと組み合わせて判定します。

運用前に確認する制約

  • ラッパー自身は、環境を絞る前に親の環境で起動します。PYTHONPATHやLD_PRELOADなどを含む親環境そのものを信頼できない場合、このラッパーだけでは起動時の影響を防げません。管理された起動元を使います。
  • ラッパーは選別のために親の環境を読み取れます。コードとポリシーは、信頼できる場所で管理します。
  • 絶対パスはPATHによる取り違えを防ぎますが、実行ファイルの改変、シンボリックリンクの差し替え、ハッシュ固定までは扱いません。
  • shebangが/usr/bin/env python3のスクリプトを直接指定すると、最小化した環境では起動できない場合があります。例のように、絶対パスのPythonとスクリプトを別々の引数にします。
  • 対象CLIが別のコマンドを起動する場合も、絶対パスを使うか、内容を確認した固定PATHを渡します。PATHがない場合の検索動作を一律に仮定しません。
  • カレントディレクトリ、umask、ユーザーID、リソース上限、ネットワーク、標準入出力、継承可能なファイル記述子は隔離しません。
  • 秘密をコマンドライン引数へ移すと、プロセス一覧などから見える可能性があります。環境変数の削減は、秘密の安全な受け渡し方法そのものではありません。
  • 実行後の記録、タイムアウト、再試行は外側の監督処理で行います。副作用のあるCLIを再試行する場合は、第174回と同様に冪等性を確認します。

今回のまとめ

親の環境を丸ごと継承せず、引き継ぐ名前と固定値を明示すると、対象CLIへ不要な設定や秘密値が渡る範囲を小さくできます。

まず確認用CLIで、必要な変数が存在し、不要な変数が存在しないことを確かめます。その後、対象CLIが実際に必要とする変数だけを追加してください。ファイルやネットワークも制限する場合は、環境変数の許可リストとは別の隔離が必要です。

第175回 Python処理のタイムアウト時にプロセスグループへ停止信号を送る

第172回では複数のCLIを順番に実行し、第173回では二重起動を防ぎ、第174回では一時失敗だけを再試行しました。

ただし、タイムアウト時に直接の子プロセスだけを終了すると、その子が起動した処理は残ることがあります。そのまま再試行すれば、新旧の処理が重なります。

今回は子コマンドを新しいプロセスグループで起動し、タイムアウト時に同じグループへSIGTERMを送ります。猶予後も残る処理にはSIGKILLを送り、直接の子を回収して終了するラッパーを作ります。

対象はPython 3.12以上を実行するLinuxなど、POSIXシグナルとos.killpg()を利用できる環境です。外部ライブラリは使いません。

プロセスグループを使う理由

Popen.terminate()が停止するのは直接の子です。子が同じプロセスグループ内に孫プロセスを起動しても、その孫には信号が届きません。

start_new_session=Trueを指定すると、POSIX環境では子の実行前にsetsid()が呼ばれます。直接の子は新しいセッションとプロセスグループのリーダーになるため、そのPIDをグループIDとしてos.killpg()へ渡せます。

ただし、別のグループやセッションへ移動した子孫には届きません。外部サービスへの送信や作成済みファイルも取り消せません。

今回の実行契約

項目 仕様
子コマンド --より後ろの引数配列を、シェルを介さず実行する
タイムアウト --timeoutで1~86400秒を指定する
終了猶予 --graceで0~60秒を指定する。既定値は5秒
標準入力 DEVNULLへ接続し、対話入力を待たせない
標準出力・標準エラー出力 子コマンドの出力先を引き継ぐ。ラッパーの診断は標準エラー出力へ書く
タイムアウト時 グループへSIGTERMを送り、猶予後にSIGKILLを送る
Ctrl+C 子の終了待ち中に受けた場合は同じ後始末を行い、成功時は130を返す
確認範囲 SIGKILL後、直接の子が5秒以内に終了したことまで確認する

--timeoutはPopen.wait()の待機時間です。プロセスの起動時間、終了猶予、回収待ちは含みません。ラッパー全体の厳密な実行時間上限ではありません。

終了コード124は、タイムアウト後に直接の子を回収できたことを表します。プロセスグループ全体の消滅を保証する値ではありません。後始末に失敗した場合は125を返します。

1.実行ラッパーを保存する

次のコードをgroup_timeout.pyとして保存します。

#!/usr/bin/env python3
import argparse
import os
import signal
import subprocess
import sys
import time

TIMEOUT_EXIT = 124
CLEANUP_EXIT = 125
CANNOT_EXECUTE_EXIT = 126
NOT_FOUND_EXIT = 127
INTERRUPTED_EXIT = 130
KILL_WAIT_SECONDS = 5


def integer_between(minimum, maximum):
    def parse(text):
        try:
            value = int(text)
        except ValueError as exc:
            raise argparse.ArgumentTypeError(
                "整数で指定してください"
            ) from exc

        if not minimum <= value <= maximum:
            raise argparse.ArgumentTypeError(
                f"{minimum}から{maximum}までの整数で指定してください"
            )
        return value

    return parse


def parse_args(argv):
    parser = argparse.ArgumentParser(
        description="子コマンドを新しいプロセスグループで実行します"
    )
    parser.add_argument(
        "--timeout",
        required=True,
        type=integer_between(1, 86400),
        help="終了待ちのタイムアウト(秒、1から86400)",
    )
    parser.add_argument(
        "--grace",
        default=5,
        type=integer_between(0, 60),
        help="SIGTERM後の猶予(秒、0から60、既定値5)",
    )
    parser.add_argument("command", nargs=argparse.REMAINDER)
    args = parser.parse_args(argv)

    if len(args.command) < 2 or args.command[0] != "--":
        parser.error("子コマンドの前に -- を置いてください")

    args.command = args.command[1:]
    return args


def report(message):
    print(
        f"group_timeout: {message}",
        file=sys.stderr,
        flush=True,
    )


def send_group(process_group_id, signal_number):
    try:
        os.killpg(process_group_id, signal_number)
    except ProcessLookupError:
        return False
    return True


def wait_grace(seconds):
    deadline = time.monotonic() + seconds
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return
        try:
            time.sleep(min(0.1, remaining))
        except KeyboardInterrupt:
            report("追加の割り込みを受けたため猶予を打ち切ります")
            return


def reap_after_kill(process):
    deadline = time.monotonic() + KILL_WAIT_SECONDS
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return False
        try:
            process.wait(timeout=remaining)
            return True
        except subprocess.TimeoutExpired:
            return False
        except KeyboardInterrupt:
            continue


def signal_group_and_reap_child(process, grace):
    process_group_id = process.pid

    if send_group(process_group_id, signal.SIGTERM):
        report(
            f"process_group={process_group_id} signal=SIGTERM"
        )
        wait_grace(grace)

    if send_group(process_group_id, signal.SIGKILL):
        report(
            f"process_group={process_group_id} signal=SIGKILL"
        )

    return reap_after_kill(process)


def cleanup(process, grace):
    try:
        reaped = signal_group_and_reap_child(process, grace)
    except OSError as exc:
        report(f"プロセスグループへの信号送信に失敗しました: {exc}")
        try:
            process.kill()
            process.wait(timeout=KILL_WAIT_SECONDS)
        except (
            OSError,
            subprocess.TimeoutExpired,
            KeyboardInterrupt,
        ):
            pass
        return False

    if not reaped:
        report(
            "SIGKILL後も直接の子プロセスの終了を確認できませんでした"
        )
    return reaped


def main(argv=None):
    args = parse_args(argv)

    if os.name != "posix" or not hasattr(os, "killpg"):
        report("os.killpgを使えるPOSIX環境専用です")
        return 2

    try:
        process = subprocess.Popen(
            args.command,
            stdin=subprocess.DEVNULL,
            stdout=None,
            stderr=None,
            shell=False,
            start_new_session=True,
        )
    except FileNotFoundError as exc:
        report(f"実行ファイルが見つかりません: {exc}")
        return NOT_FOUND_EXIT
    except PermissionError as exc:
        report(f"実行権限がありません: {exc}")
        return CANNOT_EXECUTE_EXIT
    except OSError as exc:
        report(f"子コマンドを起動できません: {exc}")
        return CANNOT_EXECUTE_EXIT

    try:
        return_code = process.wait(timeout=args.timeout)
    except subprocess.TimeoutExpired:
        report(f"timeout={args.timeout}s")
        if not cleanup(process, args.grace):
            return CLEANUP_EXIT
        return TIMEOUT_EXIT
    except KeyboardInterrupt:
        report("keyboard_interrupt")
        if not cleanup(process, args.grace):
            return CLEANUP_EXIT
        return INTERRUPTED_EXIT

    if return_code < 0:
        return 128 + (-return_code)
    return return_code


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

猶予中に直接の子が終了しても、指定時間が終わるまで待ちます。親だけがSIGTERMで終了し、同じグループの孫が残る場合にも、その後のSIGKILLを送るためです。

shell=Falseなので、ワイルドカード、パイプ、リダイレクトは解釈されません。ただし、利用者が自由にコマンド名や引数を指定できる設計にすれば、任意のプログラムや危険なオプションを実行されます。実運用では許可するコマンドを固定してください。

2.正常終了と子コマンドの失敗を確認する

まず正常終了を確認します。

python3 group_timeout.py --timeout 5 --grace 1 -- \
  python3 -c 'print("ok")'
echo $?

標準出力にはok、終了コードには0が表示されます。

子コマンドが終了コード7を返す場合も確認します。

python3 group_timeout.py --timeout 5 -- \
  python3 -c 'raise SystemExit(7)'
echo $?

終了コードは7です。子がシグナル番号Nで終了した場合は、Popenが返す負の値を128 + Nへ変換します。

3.SIGTERMを無視する孫プロセスで確認する

次のコードをstubborn_tree.pyとして保存します。孫はSIGTERMを無視します。親は、孫がシグナルハンドラの設定を終えたことを確認してからPIDを表示します。

#!/usr/bin/env python3
import os
import signal
import subprocess
import sys
import time
from pathlib import Path


if sys.argv[1:] == ["grandchild"]:
    signal.signal(signal.SIGTERM, signal.SIG_IGN)
    print("ready", flush=True)
    while True:
        time.sleep(1)

if sys.argv[1:]:
    raise SystemExit("usage: python3 stubborn_tree.py")

grandchild = subprocess.Popen(
    [
        sys.executable,
        str(Path(__file__).resolve()),
        "grandchild",
    ],
    stdin=subprocess.DEVNULL,
    stdout=subprocess.PIPE,
    text=True,
)

ready_stream = grandchild.stdout
if ready_stream is None:
    raise RuntimeError("grandchild stdout is unavailable")

ready = ready_stream.readline()
ready_stream.close()
if ready != "ready\n":
    return_code = grandchild.wait()
    raise SystemExit(
        f"grandchild setup failed: return_code={return_code}"
    )

print(
    f"parent_pid={os.getpid()} grandchild_pid={grandchild.pid}",
    flush=True,
)

while True:
    time.sleep(1)

2秒でタイムアウトさせ、SIGTERMから1秒後にSIGKILLへ進めます。

python3 group_timeout.py --timeout 2 --grace 1 -- \
  python3 stubborn_tree.py
echo $?

出力はおおむね次の形になります。PIDは実行ごとに変わります。

parent_pid=<親PID> grandchild_pid=<孫PID>
group_timeout: timeout=2s
group_timeout: process_group=<親PID> signal=SIGTERM
group_timeout: process_group=<親PID> signal=SIGKILL
124

124で確認できるのは直接の子の回収までです。表示されたPIDを使い、直後にプロセス表も調べます。

ps -o pid=,ppid=,pgid=,stat=,args= \
  -p <親PID>,<孫PID>

何も表示されなければ、両方ともプロセス表から消えています。状態欄がZなら実行は終わっていますが、回収待ちです。それ以外の状態で対象のコマンドが残っている場合は、自動再試行せず、PGIDと起動方法を確認します。

4.終了コードを運用へ渡す

終了コード 意味
0 子コマンドが正常終了した
子コマンドの非ゼロ値 通常終了した子の値をそのまま返した
2 引数が不正、または対象外の環境だった
124 タイムアウト後、直接の子の終了を確認した
125 信号送信に失敗したか、直接の子の終了を確認できなかった
126 権限などの理由で子コマンドを起動できなかった
127 実行ファイルが見つからなかった
130 終了待ち中にCtrl+Cを受け、後始末を完了した
128 + N 子がシグナル番号Nで終了した

存在しない実行ファイルは127になります。

python3 group_timeout.py --timeout 5 -- \
  ./does-not-exist-for-test
echo $?

--timeout 0、範囲外の--grace、子コマンド前の--がない呼び出しは、実行前に引数エラーとなります。

124~127や130は子コマンド自身も返せます。終了コードだけで原因を判定する運用では、これらをラッパー用として子側の契約から除外してください。ラッパーが検出した事象には、標準エラー出力へgroup_timeout:で始まる診断も出ます。

このラッパーが扱えない範囲

  • 子孫がsetsid()やsetpgid()で別のグループへ移動すると、今回のkillpg()では停止できない
  • 別ホスト、コンテナ管理機能、ジョブキューへ渡した処理は対象外である
  • 直接の子が正常終了し、バックグラウンド処理だけを残した場合は停止処理を行わない
  • SIGKILLでは終了処理を実行できないため、一時ファイルや未完成の出力が残り得る
  • 外部APIへの登録、メール送信、確定済みデータなどの副作用は取り消さない
  • ラッパー自身へのSIGTERM、SIGHUP、SIGKILL、電源断の後始末は実装していない
  • 起動中や後始末中に繰り返し届くSIGINTを、あらゆる位置で処理する設計ではない

子コマンドは、自分が起動した処理を待ってから正常終了する設計にします。出力は一時ファイルへ書き、検証後に置き換えるなど、途中終了を前提にしてください。

再試行と組み合わせる順序

  1. ワークフロー全体の排他ロックを取得する
  2. 再試行ラッパーが1回分の試行を始める
  3. 今回のラッパーが、その1回にタイムアウトを設定する
  4. 実際の子CLIを新しいプロセスグループで起動する

終了コード124を無条件に再試行対象へ加えてはいけません。外部更新の直後に応答待ちで止まった可能性があるためです。処理IDなどによる重複防止があり、同じ入力を再実行しても結果が壊れない場合に限って再試行します。

このラッパーが管理するのは、タイムアウト時に信号を送る範囲と、直接の子の回収です。プロセスグループ全体の消滅、処理結果の取り消し、再実行の安全性は別に確認します。

第174回 失敗したPython処理を無条件に再実行しない:終了コードで制御する再試行ラッパー

第172回では複数のPython CLIを順番に実行し、第173回ではワークフローの二重起動を排他ロックで防ぎました。それでも、一時的な資源不足や接続失敗で子コマンドが止まることはあります。

ここで、すべての失敗を同じように再実行するのは危険です。外部サービスへの登録が完了した直後に応答だけ受け取れなかった場合、再実行によって同じ登録や通知が繰り返される可能性があります。

今回は、子コマンド側で「一時失敗」と定義した終了コードだけを、上限回数と待機時間を決めて再試行するラッパーを作ります。対象はPython 3.10以上のLinuxです。外部ライブラリは使いません。

排他制御と再試行は解決する問題が異なる

仕組み 防ぐ問題 防げない問題
順次実行 処理順の乱れ、失敗後の後続実行 同じ処理の二重起動
排他ロック 同じホスト上での同時実行 時間をずらした重複処理
今回の再試行 明示された一時失敗からの再実行 副作用の重複、処理の冪等性

サンプルでは終了コード75を一時失敗として扱います。ただし、75がすべてのCLIで共通して一時失敗を意味するわけではありません。再試行するコードは、子コマンドとの入出力契約で先に決めます。

今回の実行契約

項目 仕様
子コマンド --より後ろの引数配列を、シェルを介さず実行する
再試行対象 --retry-codeで明示した1~124の終了コードだけ
実行回数 初回を含め1~10回。既定値は3回
待機 初期待機時間に係数を掛け、上限値で打ち止めにする
標準入力 DEVNULLへ接続し、対話入力を待たせない
子の出力 試行ごとに一時ファイルへ保存し、実行中は表示しない
最終成功 成功した試行の標準出力と標準エラー出力だけを返す
最終失敗 標準出力は捨て、各失敗のメタデータと標準エラー出力を返す
環境変数 試行番号と最大回数を子へ渡す

途中の失敗後に成功した場合、以前の失敗ログは残しません。最終成功時の出力を後続処理へそのまま渡すための仕様です。全試行の監査記録が必要な業務では、この実装とは別に、アクセス制御したログ保存先を用意してください。

再試行ラッパーを作る

次のコードをBOMなしUTF-8のretry_command.pyとして保存します。

#!/usr/bin/env python3
import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time


def bounded_int(label, low, high):
    def parse(text):
        try:
            value = int(text, 10)
        except ValueError as exc:
            raise argparse.ArgumentTypeError(
                f"{label}は整数で指定してください"
            ) from exc
        if not low <= value <= high:
            raise argparse.ArgumentTypeError(
                f"{label}は{low}以上{high}以下で指定してください"
            )
        return value

    return parse


def parse_args():
    parser = argparse.ArgumentParser(
        description="指定した終了コードだけを再試行する"
    )
    parser.add_argument(
        "--retry-code",
        action="append",
        required=True,
        type=bounded_int("retry-code", 1, 124),
    )
    parser.add_argument(
        "--max-attempts",
        type=bounded_int("max-attempts", 1, 10),
        default=3,
    )
    parser.add_argument(
        "--initial-delay-ms",
        type=bounded_int("initial-delay-ms", 0, 60_000),
        default=1_000,
    )
    parser.add_argument(
        "--backoff-factor",
        type=bounded_int("backoff-factor", 1, 10),
        default=2,
    )
    parser.add_argument(
        "--max-delay-ms",
        type=bounded_int("max-delay-ms", 0, 60_000),
        default=30_000,
    )
    parser.add_argument("command", nargs=argparse.REMAINDER)
    args = parser.parse_args()

    if not args.command or args.command[0] != "--":
        parser.error("子コマンドの前に -- を指定してください")
    args.command = args.command[1:]
    if not args.command:
        parser.error("-- の後に子コマンドが必要です")
    if args.initial_delay_ms > args.max_delay_ms:
        parser.error("initial-delay-msはmax-delay-ms以下にしてください")

    args.retry_codes = frozenset(args.retry_code)
    return args


def append_record(history, payload, child_stderr=None):
    line = json.dumps(
        payload, ensure_ascii=False, separators=(",", ":")
    )
    history.write((line + "\n").encode("utf-8"))

    if child_stderr is None:
        return
    child_stderr.seek(0, os.SEEK_END)
    if child_stderr.tell() == 0:
        return

    child_stderr.seek(0)
    shutil.copyfileobj(child_stderr, history, length=1024 * 1024)
    child_stderr.seek(-1, os.SEEK_END)
    if child_stderr.read(1) != b"\n":
        history.write(b"\n")


def replay(source, target):
    source.seek(0)
    shutil.copyfileobj(source, target, length=1024 * 1024)
    target.flush()


def normalized_status(returncode):
    if returncode >= 0:
        return min(returncode, 255)
    return min(128 + (-returncode), 255)


def run(args):
    delay_ms = args.initial_delay_ms

    with tempfile.TemporaryFile() as history:
        for attempt in range(1, args.max_attempts + 1):
            environment = os.environ.copy()
            environment["RETRY_WRAPPER_ATTEMPT"] = str(attempt)
            environment["RETRY_WRAPPER_MAX_ATTEMPTS"] = str(
                args.max_attempts
            )

            with (
                tempfile.TemporaryFile() as child_stdout,
                tempfile.TemporaryFile() as child_stderr,
            ):
                try:
                    completed = subprocess.run(
                        args.command,
                        stdin=subprocess.DEVNULL,
                        stdout=child_stdout,
                        stderr=child_stderr,
                        shell=False,
                        check=False,
                        env=environment,
                    )
                except OSError as exc:
                    if isinstance(exc, FileNotFoundError):
                        kind, status = "not_found", 127
                    elif isinstance(exc, PermissionError):
                        kind, status = "permission_denied", 126
                    else:
                        kind, status = "os_error", 125
                    append_record(
                        history,
                        {
                            "source": "retry_wrapper",
                            "event": "launch_error",
                            "attempt": attempt,
                            "max_attempts": args.max_attempts,
                            "kind": kind,
                            "retrying": False,
                        },
                    )
                    replay(history, sys.stderr.buffer)
                    return status

                returncode = completed.returncode
                if returncode == 0:
                    replay(child_stdout, sys.stdout.buffer)
                    replay(child_stderr, sys.stderr.buffer)
                    return 0

                retrying = (
                    returncode in args.retry_codes
                    and attempt < args.max_attempts
                )
                event = {
                    "source": "retry_wrapper",
                    "event": "attempt_failed",
                    "attempt": attempt,
                    "max_attempts": args.max_attempts,
                    "returncode": returncode,
                    "retrying": retrying,
                }
                if retrying:
                    event["next_delay_ms"] = delay_ms
                append_record(history, event, child_stderr)

                if not retrying:
                    replay(history, sys.stderr.buffer)
                    return normalized_status(returncode)

            time.sleep(delay_ms / 1000)
            delay_ms = min(
                args.max_delay_ms,
                delay_ms * args.backoff_factor,
            )

    raise AssertionError("到達しない分岐です")


def wrapper_event(event, kind):
    payload = {
        "source": "retry_wrapper",
        "event": event,
        "kind": kind,
    }
    print(
        json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
        file=sys.stderr,
        flush=True,
    )


def main():
    args = parse_args()
    try:
        return run(args)
    except KeyboardInterrupt:
        wrapper_event("interrupted", "keyboard_interrupt")
        return 130
    except BrokenPipeError:
        try:
            sys.stdout.close()
        except BrokenPipeError:
            pass
        return 141
    except OSError as exc:
        wrapper_event("wrapper_error", type(exc).__name__)
        return 125


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

処理の順序を確認する

  1. 子コマンドを一時ファイルへ接続して起動する
  2. 終了コード0なら、その試行の出力だけを呼び出し元へ返す
  3. 非ゼロなら、終了コードが再試行リストにあるか確認する
  4. 再試行する場合は待機し、次の待機時間を係数倍する
  5. 対象外の終了コードまたは上限到達なら、失敗履歴を標準エラー出力へ返す
  6. シグナルで終了した子の負の戻り値は、このラッパーの契約として128 + シグナル番号へ変換する

Pythonの公式文書でも、POSIXでシグナル終了した子プロセスの戻り値は負のシグナル番号になると説明されています。詳細はPython 3.10のsubprocess文書で確認できます。

再試行のたびに、子コマンドへRETRY_WRAPPER_ATTEMPTとRETRY_WRAPPER_MAX_ATTEMPTSを渡します。試行番号は診断には使えますが、処理の重複を防ぐ識別子ではありません。副作用のある処理には、全試行で同じ値を使う業務上の処理IDが別に必要です。

3回目に成功する子コマンドで確認する

次のコードをdemo_job.pyとして保存します。実運用の一時障害を再現するものではなく、試行番号に応じて終了コードを変える動作確認用コードです。

#!/usr/bin/env python3
import json
import os
import sys

attempt = int(os.environ["RETRY_WRAPPER_ATTEMPT"])
if attempt < 3:
    print(f"一時失敗: attempt={attempt}", file=sys.stderr)
    raise SystemExit(75)

print(
    json.dumps(
        {"status": "ok", "attempt": attempt},
        ensure_ascii=False,
        separators=(",", ":"),
    )
)

同じディレクトリで次の順に実行します。

python3 retry_command.py --retry-code 75 --max-attempts 3 --initial-delay-ms 10 --backoff-factor 2 --max-delay-ms 20 -- python3 demo_job.py > result.json 2> retry.err
status=$?
printf 'exit=%s\n' "$status"
cat result.json
wc -c < retry.err

結果は次のとおりです。

exit=0
{"status":"ok","attempt":3}
0

1回目の後に10ミリ秒、2回目の後に20ミリ秒待機します。3回目が成功すると、それ以前の標準出力、標準エラー出力、失敗イベントは破棄されます。

恒久エラーは1回で止める

終了コード65を再試行リストへ含めていないため、次のコマンドは1回で停止します。終了コードを確認する行は、コマンドの直後に実行してください。

python3 retry_command.py --retry-code 75 --initial-delay-ms 0 --max-delay-ms 0 -- python3 -c 'import sys; print("設定エラー", file=sys.stderr); raise SystemExit(65)' > permanent.out 2> permanent.err
status=$?
printf 'exit=%s\n' "$status"
cat permanent.err

permanent.outは空です。終了コードと標準エラー出力は次のようになります。

exit=65
{"source":"retry_wrapper","event":"attempt_failed","attempt":1,"max_attempts":3,"returncode":65,"retrying":false}
設定エラー

終了コードを確認する

終了経路 ラッパーの終了コード 再試行
子が成功 0 終了する
対象外の子終了コード65 65 しない
対象コード75で上限到達 75 それ以上しない
実行ファイルが存在しない 127 しない
実行権限がない 126 しない
ラッパー内のOSエラー 125 しない
子がSIGTERMで終了 143 しない
Ctrl+C 130 しない
出力先のパイプが閉じられた 141 しない
引数エラー 2 起動前に停止する

子がSIGTERMで終了すると、subprocessの戻り値は-15です。失敗イベントには-15を記録し、ラッパー自体は143で終了します。シグナル終了は通常の一時失敗として再試行しません。

再試行の前に冪等性を確認する

再試行できるのは、同じ処理を繰り返しても業務上の結果が重複しない場合です。少なくとも、次のいずれかが必要です。

  • 処理が読み取り専用である
  • 同じ処理IDによる登録を保存先が拒否する
  • 再実行前に、すでに完了しているか照会できる
  • 一時ファイルへ出力し、成功時にだけ確定できる

メール送信、請求登録、外部APIの更新、追記型ファイルへの書き込みなどは、排他ロックだけでは重複を防げません。外部処理が完了した後で応答だけ失われた場合、終了コードからは「未実行」と「実行済み」を区別できないためです。

複数ステップからなるワークフロー全体を再試行すると、失敗前に完了したステップまで繰り返されます。再試行ラッパーは、再実行可能だと確認した個別ステップへ適用します。第173回の排他ロックと組み合わせる場合は、ロックの内側で全試行を実行し、待機中も同じロックを保持します。

この小さな実装の制約

  • タイムアウトは実装していないため、終了しない子コマンドを待ち続ける
  • 待機時間にランダムな揺らぎがなく、多数の端末が同時実行すると再試行時刻も重なり得る
  • 子が起動した孫プロセスの終了や、ラッパー自身がSIGTERMを受けた場合の転送は管理しない
  • 書き込み可能で十分な空き容量のあるOSの一時領域が必要になる
  • 子の出力サイズを制限しておらず、システムの一時領域を消費する
  • 出力は子の終了後に再生されるため、実行中のログをリアルタイムには確認できない
  • 標準出力と標準エラー出力を別々に保存するため、両者の細かな出力順序は再現しない
  • 最終成功時には、それ以前の失敗履歴を保存しない

この実装の対象は、出力サイズと実行時間を別の運用ルールで制限できる、ローカルの小規模なバッチ処理です。まず既存CLIを一つ選び、成功、恒久失敗、一時失敗の終了コードを定義します。そのうえで、副作用の重複を防げる処理だけを再試行対象にしてください。

第173回 Pythonワークフローの二重起動を防ぐ:fcntl.flockで作る排他実行ラッパー

第172回では、複数のPython CLIを設定順に実行し、失敗した時点で後続処理を止めました。ただし、手動実行と定期実行が重なると、ワークフロー全体が二つ同時に動く可能性があります。

今回は、同じLinuxホスト上で、同じロックを使うコマンドを一つだけ実行できるラッパーを作ります。Python標準ライブラリのfcntl.flock()で非待機ロックを取得し、すでに実行中なら子コマンドを起動せず、終了コード75で停止します。

この仕組みが防ぐのは同時実行です。完了後の再実行、外部APIへの二重送信、途中まで作成されたファイルの取り消しは扱いません。

順番制御と排他制御は別に考える

仕組み 防ぐ問題 防げない問題
第172回の順次実行器 実行順の乱れ、失敗後の後続実行 ワークフロー全体の二重起動
今回の排他ラッパー 同じロックを使うコマンドの同時実行 完了後の再実行、ラッパーを通さない実行
業務側の重複防止 同じ処理IDの再登録や二重送信 異なる処理同士の資源競合

請求、通知、外部API更新などには、処理IDを使った重複防止も別に必要です。

今回の実行契約

項目 仕様
動作環境 Python 3.11以上、Linux、ローカルファイルシステム
ロック 指定ファイルへ排他的な非待機ロックを取得する
競合時 子コマンドを起動せず、標準エラー出力へ1行JSONを出して終了コード75
子コマンド シェルを介さず、--より後ろの引数配列を実行する
標準入力 DEVNULLへ接続し、対話入力を待たせない
標準出力・標準エラー出力 子コマンドの出力先をそのまま引き継ぐ
終了コード 子コマンドの値を引き継ぐ。シグナル終了は128 + シグナル番号へ変換する
ロックファイル 終了後も削除しない。ファイルの存在ではなく、カーネルが管理するロック状態で判定する
対象外 ネットワークファイルシステム、複数ホストの分散ロック、タイムアウト、再試行、ロールバック

flock()のロックは協調型です。このラッパーを通さない処理までは止められません。非待機ロックとpass_fdsの動作は、fcntl.flockの公式ドキュメントとsubprocess.Popenの公式ドキュメントで確認できます。

1.排他実行ラッパーを作る

次のコードをsingle_run.pyとして保存します。ロックファイルの親ディレクトリは事前に作成し、そのディレクトリへ書き込める利用者を限定してください。

from __future__ import annotations

import argparse
import errno
import fcntl
import json
import os
import stat
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import NoReturn, TextIO


EXIT_USAGE = 2
EXIT_RUNTIME = 3
EXIT_BUSY = 75


def emit_json(stream: TextIO, payload: dict[str, object]) -> None:
    stream.write(
        json.dumps(payload, ensure_ascii=False, separators=(',', ':')) + '\n'
    )
    stream.flush()


class AppError(Exception):
    def __init__(self, exit_code: int, payload: dict[str, object]) -> None:
        super().__init__(str(payload.get('reason', 'error')))
        self.exit_code = exit_code
        self.payload = payload


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> NoReturn:
        emit_json(
            sys.stderr,
            {
                'status': 'error',
                'reason': 'invalid_arguments',
                'message': message,
            },
        )
        raise SystemExit(EXIT_USAGE)


def parse_args(
    argv: list[str] | None = None,
) -> tuple[Path, str, list[str]]:
    raw = list(sys.argv[1:] if argv is None else argv)
    parser = JsonArgumentParser(
        usage='%(prog)s --lock LOCK_FILE -- COMMAND [ARG ...]',
        description='同じロックを使うコマンドの同時実行を防ぎます。',
    )
    parser.add_argument('--lock', required=True, type=Path)

    if '--' not in raw:
        if '-h' in raw or '--help' in raw:
            parser.parse_args(raw)
        parser.error('区切りの -- と実行コマンドが必要です')

    separator = raw.index('--')
    namespace = parser.parse_args(raw[:separator])
    command = raw[separator + 1:]
    if not command:
        parser.error('-- の後に実行コマンドを指定してください')

    display_path = os.fspath(namespace.lock)
    return namespace.lock, display_path, command


def write_lock_record(lock_fd: int, record: dict[str, object]) -> None:
    data = (
        json.dumps(record, ensure_ascii=False, separators=(',', ':')) + '\n'
    ).encode('utf-8')

    os.ftruncate(lock_fd, 0)
    os.lseek(lock_fd, 0, os.SEEK_SET)
    remaining = memoryview(data)
    while remaining:
        written = os.write(lock_fd, remaining)
        if written == 0:
            raise OSError(errno.EIO, 'lock metadata write returned zero bytes')
        remaining = remaining[written:]
    os.fsync(lock_fd)


def acquire_lock(lock_path: Path, display_path: str) -> int:
    flags = os.O_RDWR | os.O_CREAT
    flags |= getattr(os, 'O_CLOEXEC', 0)
    flags |= getattr(os, 'O_NOFOLLOW', 0)

    try:
        lock_fd = os.open(lock_path, flags, 0o600)
    except FileNotFoundError as exc:
        raise AppError(
            EXIT_RUNTIME,
            {
                'status': 'error',
                'reason': 'lock_parent_not_found',
                'lock': display_path,
            },
        ) from exc
    except PermissionError as exc:
        raise AppError(
            EXIT_RUNTIME,
            {
                'status': 'error',
                'reason': 'lock_permission_denied',
                'lock': display_path,
            },
        ) from exc
    except OSError as exc:
        raise AppError(
            EXIT_RUNTIME,
            {
                'status': 'error',
                'reason': 'lock_open_failed',
                'lock': display_path,
            },
        ) from exc

    try:
        try:
            mode = os.fstat(lock_fd).st_mode
        except OSError as exc:
            raise AppError(
                EXIT_RUNTIME,
                {
                    'status': 'error',
                    'reason': 'lock_stat_failed',
                    'lock': display_path,
                },
            ) from exc

        if not stat.S_ISREG(mode):
            raise AppError(
                EXIT_RUNTIME,
                {
                    'status': 'error',
                    'reason': 'lock_not_regular_file',
                    'lock': display_path,
                },
            )

        try:
            fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError as exc:
            busy_numbers = {
                errno.EACCES,
                errno.EAGAIN,
                errno.EWOULDBLOCK,
            }
            if exc.errno in busy_numbers:
                raise AppError(
                    EXIT_BUSY,
                    {
                        'status': 'busy',
                        'reason': 'lock_held',
                        'lock': display_path,
                    },
                ) from exc
            raise AppError(
                EXIT_RUNTIME,
                {
                    'status': 'error',
                    'reason': 'lock_acquire_failed',
                    'lock': display_path,
                },
            ) from exc

        record = {
            'format_version': 1,
            'owner_pid': os.getpid(),
            'acquired_at_utc': datetime.now(timezone.utc)
            .isoformat(timespec='seconds')
            .replace('+00:00', 'Z'),
        }
        try:
            write_lock_record(lock_fd, record)
        except OSError as exc:
            raise AppError(
                EXIT_RUNTIME,
                {
                    'status': 'error',
                    'reason': 'lock_metadata_write_failed',
                    'lock': display_path,
                },
            ) from exc

        return lock_fd
    except BaseException:
        try:
            os.close(lock_fd)
        except OSError:
            pass
        raise


def normalize_returncode(returncode: int) -> int:
    if returncode >= 0:
        return returncode
    signal_number = -returncode
    return min(255, 128 + signal_number)


def run_child(command: list[str], lock_fd: int) -> int:
    try:
        completed = subprocess.run(
            command,
            stdin=subprocess.DEVNULL,
            check=False,
            shell=False,
            close_fds=True,
            pass_fds=(lock_fd,),
        )
    except FileNotFoundError as exc:
        raise AppError(
            EXIT_RUNTIME,
            {
                'status': 'error',
                'reason': 'command_not_found',
                'command': command[0],
            },
        ) from exc
    except PermissionError as exc:
        raise AppError(
            EXIT_RUNTIME,
            {
                'status': 'error',
                'reason': 'command_not_executable',
                'command': command[0],
            },
        ) from exc
    except OSError as exc:
        raise AppError(
            EXIT_RUNTIME,
            {
                'status': 'error',
                'reason': 'child_start_failed',
                'command': command[0],
            },
        ) from exc

    return normalize_returncode(completed.returncode)


def close_lock(lock_fd: int) -> None:
    # 明示的に解錠せず、複製された記述子がすべて閉じた時点で解放する。
    try:
        os.close(lock_fd)
    except OSError:
        pass


def main(argv: list[str] | None = None) -> int:
    lock_fd: int | None = None
    try:
        lock_arg, display_path, command = parse_args(argv)
        lock_path = Path(os.path.abspath(os.fspath(lock_arg)))
        lock_fd = acquire_lock(lock_path, display_path)
        return run_child(command, lock_fd)
    except AppError as exc:
        emit_json(sys.stderr, exc.payload)
        return exc.exit_code
    except KeyboardInterrupt:
        return 130
    except OSError:
        emit_json(
            sys.stderr,
            {'status': 'error', 'reason': 'unexpected_os_error'},
        )
        return EXIT_RUNTIME
    finally:
        if lock_fd is not None:
            close_lock(lock_fd)


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

pass_fdsでロックのファイル記述子を直接の子プロセスにも渡しています。Linuxのflock()は複製された記述子と同じロックを共有し、最後の記述子が閉じるまで保持されます。このため、ラッパーだけが強制終了しても、子プロセスが動いている間はロックが残ります。

反対に、子プロセスが継承した記述子へ明示的なLOCK_UNを実行すると、ラッパーが持つロックも解除されます。子プロセスからこの記述子を操作しないことが前提です。

ロックファイルにはコマンドや引数を書きません。引数に秘密情報が含まれる場合があるためです。PIDと取得時刻は調査用であり、現在も実行中であることの証明には使いません。

2.動作確認用の子コマンドを作る

次のコードをdemo_job.pyとして、single_run.pyと同じディレクトリへ保存します。

from __future__ import annotations

import argparse
import json
import time


def seconds_arg(text: str) -> int:
    try:
        value = int(text)
    except ValueError as exc:
        raise argparse.ArgumentTypeError('整数を指定してください') from exc
    if not 1 <= value <= 30:
        raise argparse.ArgumentTypeError('1から30までを指定してください')
    return value


parser = argparse.ArgumentParser()
parser.add_argument('--seconds', required=True, type=seconds_arg)
args = parser.parse_args()

time.sleep(args.seconds)
print(
    json.dumps(
        {'status': 'ok', 'slept_seconds': args.seconds},
        ensure_ascii=False,
        separators=(',', ':'),
    )
)

ロックを取得し、子コマンドを8秒間動かします。

python3 single_run.py --lock workflow.lock -- python3 demo_job.py --seconds 8

完了すると、子コマンドの標準出力が表示されます。

{"status":"ok","slept_seconds":8}

workflow.lockは初回に作成され、処理後も残ります。正常に記録された内容は次のコマンドで確認できます。

python3 -m json.tool workflow.lock

メタデータの書き込み中にプロセスを強制終了すると、内容が空または不完全になる可能性があります。次回のロック取得時に上書きされるため、内容ではなくロック取得の成否を実行中の判定に使います。

3.二つのシェルから同時に実行する

一つ目のシェルで8秒の処理を開始し、完了する前に二つ目のシェルで次を実行します。

python3 single_run.py --lock workflow.lock -- python3 demo_job.py --seconds 1

二つ目は待機せず、標準エラー出力へ次のJSONを出します。

{"status":"busy","reason":"lock_held","lock":"workflow.lock"}

終了コードは75です。

echo $?
75

一つ目が完了した後なら、ロックファイルが残っていても正常に開始できます。

4.子コマンドの失敗を引き継ぐ

子コマンドが終了コード7を返す例です。

python3 single_run.py --lock workflow.lock -- python3 -c 'raise SystemExit(7)'
echo $?
7

存在しない実行ファイルを指定した場合は、ラッパー自身のエラーとして終了コード3になります。

python3 single_run.py --lock workflow.lock -- ./not-present-command
{"status":"error","reason":"command_not_found","command":"./not-present-command"}
終了コード 意味
0 子コマンドが正常終了
2 ラッパーの引数が不正
3 ロックファイル操作または子コマンド起動に失敗
75 指定したロックを別の処理が保持中
130 ラッパーがKeyboardInterruptを受けた
そのほか 子コマンドの終了コード、またはシグナルを変換した値

子コマンド自身も2、3、75を返す可能性があります。ラッパー側の失敗かを区別する場合は、終了コードだけでなく、標準エラー出力のstatusとreasonも確認します。

通常の端末でCtrl+Cを押すと、同じフォアグラウンドプロセスグループにいるラッパーと子プロセスの両方へSIGINTが送られます。このコードに独自のシグナル転送処理はありません。ただし、subprocess.run()が子プロセスの終了を待っている間にラッパーだけがKeyboardInterruptを受けた場合は、subprocess.run()の例外処理が直接の子プロセスを強制終了してから例外を再送出し、main()が終了コード130を返します。そのため、直接の子プロセスがそのまま動き続けるのが通常の挙動ではありません。子プロセスが別のプロセスを起動し、その子孫にもロック記述子を継承させていた場合は、その子孫が動き続けてロックを保持する可能性があります。

第172回のワークフローへ適用する

第172回の実行コマンド全体を、今回の--より後ろへ置きます。正常時にはラッパーが出力を追加しないため、子CLIの標準出力と終了コードをそのまま受け取れます。

定期実行では、作業ディレクトリの違いによって別のロックファイルを作らないよう、ラッパー、ロックファイル、子コマンド、設定ファイルを絶対パスで指定します。複数のワークフローが同じ出力先を更新するなら、スクリプト名が異なっても同じロックファイルを使います。

運用時の制約

  • ロックファイルのパスを固定し、すべての起動経路で同じ値を使う
  • ロックファイルを削除、移動、ローテーションしない
  • PIDや取得時刻を見て「古いロック」と判断し、自動削除しない
  • 子コマンドはフォアグラウンドで動かし、別プロセスを残して先に終了しない
  • 子コマンドから継承したロック記述子を操作しない
  • NFSやSMBなどのネットワークファイルシステム、複数ホストの排他には使わない
  • パイプ、リダイレクト、ワイルドカードは解釈されない。必要な処理は個別の引数またはPython側で組み立てる

実行中にロックファイルを削除すると、既存処理は削除前のinodeを保持したまま、新しい処理が同名の別ファイルを作成できます。その結果、二つの処理が別々のファイルをロックして同時に動くことがあります。

最初に二つのシェルで競合時の動作を確認します。その後、第172回のワークフローを外側から包み、手動実行と定期実行の両方で同じ絶対パスのロックを指定します。

第172回 複数のPython処理を順番どおりに実行する:失敗時に止まる小さなワークフロー

これまで作ってきた検査や変換のCLIは、終了コードを確認して初めて次の処理へ進めます。複数のコマンドを手作業で実行すると、順番の取り違えや、失敗を見落として後続処理を動かす事故が起こり得ます。

今回は、JSONに記載したPython CLIを上から順に実行し、失敗、出力形式の違反、タイムアウトが起きた時点で停止するワークフロー実行器を作ります。Python 3.10以上の標準ライブラリだけを使い、シェルは介しません。

この実装は、各処理を一つのトランザクションにするものではありません。途中で止まっても、完了済みの処理が作成したファイルや外部システムへの変更は戻りません。各CLIは元データを直接更新せず、新しい出力を作る設計にしておく必要があります。

今回固定する実行契約

項目 仕様
動作環境 Python 3.10以上、ローカルのPOSIX環境
設定 BOMなしUTF-8のJSON。最大262,144バイト、最大20ステップ
実行順 stepsの記載順。並列実行はしない
作業ディレクトリ シンボリックリンク解決後の設定JSONが置かれたディレクトリ
成功条件 子CLIの終了コードが0、標準エラー出力が空、標準出力が1行のJSONオブジェクト
停止条件 子CLIの非ゼロ終了、出力契約違反、起動失敗、タイムアウト、割り込み
レポート 実行前には存在しないディレクトリへreport.jsonを作る
対象外 完了済み処理の取り消し、定期実行、並列処理、別端末への分散実行

この実装ではshell=Falseを指定するため、argvの要素はシェル展開されません。ただし、argvからシェル自体を明示的に起動すれば、そのシェルは構文を解釈します。設定には任意の実行ファイルを指定できるため、信頼できない相手から受け取った設定は実行しないでください。

1.子CLIの出力契約を確認する

最初に、実行器の動作確認専用CLIを用意します。次のコードをsample_gate.pyとして保存します。

from __future__ import annotations

import argparse
import json
import sys


def compact(value: object) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(',', ':'))


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument('step')
    parser.add_argument('--fail', action='store_true')
    args = parser.parse_args()

    if args.fail:
        print(
            compact({
                'status': 'failed',
                'step': args.step,
                'error': 'requested_failure',
            }),
            file=sys.stderr,
        )
        return 7

    print(compact({'status': 'passed', 'step': args.step}))
    return 0


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

通常時は標準出力へ1行のJSONを書き、終了コード0を返します。--failを付けた場合は標準エラー出力へJSONを書き、終了コード7を返します。これは業務判定を模したものではなく、停止動作を確認するためのテスト用CLIです。

2.実行順をJSONで定義する

次の内容をworkflow.jsonとして、sample_gate.pyと同じディレクトリへ保存します。

{
  "version": 1,
  "steps": [
    {
      "id": "input_check",
      "argv": ["{python}", "sample_gate.py", "input_check"],
      "timeout_seconds": 10
    },
    {
      "id": "candidate_check",
      "argv": ["{python}", "sample_gate.py", "candidate_check"],
      "timeout_seconds": 10
    },
    {
      "id": "publish_check",
      "argv": ["{python}", "sample_gate.py", "publish_check"],
      "timeout_seconds": 10
    }
  ]
}

{python}は、実行器自身を動かしているPythonのパスへ置き換えます。項目全体が{python}と一致した場合だけ置き換えます。

argvはコマンドを一つの文字列にせず、実行ファイルと引数を別々の要素にします。パイプ、リダイレクト、ワイルドカードなどのシェル構文は使えません。

3.ワークフロー実行器を作る

次のコードをworkflow_runner.pyとして保存します。

from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
import signal
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import BinaryIO

MAX_CONFIG_BYTES = 262_144
MAX_CAPTURE_BYTES = 65_536
MAX_STEPS = 20
ID_PATTERN = re.compile(r'[a-z][a-z0-9_-]{0,31}')


@dataclass(frozen=True)
class Step:
    id: str
    argv: tuple[str, ...]
    timeout_seconds: int


class ConfigError(Exception):
    pass


class StepFailure(Exception):
    def __init__(
        self,
        exit_code: int,
        kind: str,
        message: str,
        returncode: int | None,
        duration_ms: int,
        stdout_meta: dict[str, object],
        stderr_meta: dict[str, object],
    ) -> None:
        super().__init__(message)
        self.exit_code = exit_code
        self.kind = kind
        self.message = message
        self.returncode = returncode
        self.duration_ms = duration_ms
        self.stdout_meta = stdout_meta
        self.stderr_meta = stderr_meta

    def to_record(self, step_id: str) -> dict[str, object]:
        return {
            'id': step_id,
            'status': 'interrupted' if self.exit_code == 130 else 'failed',
            'kind': self.kind,
            'message': self.message,
            'returncode': self.returncode,
            'duration_ms': self.duration_ms,
            'stdout': self.stdout_meta,
            'stderr': self.stderr_meta,
        }


class CliParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        raise ConfigError(message)


def compact(value: object) -> str:
    return json.dumps(
        value,
        ensure_ascii=True,
        allow_nan=False,
        separators=(',', ':'),
    )


def emit_error(kind: str, message: str) -> None:
    print(
        compact({'status': 'error', 'kind': kind, 'message': message}),
        file=sys.stderr,
    )


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat(
        timespec='seconds'
    ).replace('+00:00', 'Z')


def elapsed_ms(started_ns: int) -> int:
    return (time.monotonic_ns() - started_ns) // 1_000_000


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


def reject_constant(value: str) -> None:
    raise ValueError(f'non-standard JSON constant: {value}')


def parse_json_float(value: str) -> float:
    result = float(value)
    if not math.isfinite(result):
        raise ValueError(f'JSON number is outside the finite range: {value}')
    return result


def parse_json_text(text: str) -> object:
    return json.loads(
        text,
        object_pairs_hook=object_from_pairs,
        parse_constant=reject_constant,
        parse_float=parse_json_float,
    )


def require_keys(
    value: object,
    expected: set[str],
    label: str,
) -> dict[str, object]:
    if type(value) is not dict:
        raise ConfigError(f'{label} must be an object')

    actual = set(value)
    missing = sorted(expected - actual)
    unknown = sorted(actual - expected)
    if missing or unknown:
        parts: list[str] = []
        if missing:
            parts.append(f'missing keys: {missing}')
        if unknown:
            parts.append(f'unknown keys: {unknown}')
        raise ConfigError(f'{label}: ' + ', '.join(parts))
    return value


def load_workflow(path: Path) -> tuple[tuple[Step, ...], str]:
    raw = path.read_bytes()
    if len(raw) > MAX_CONFIG_BYTES:
        raise ConfigError('workflow file is too large')
    if raw.startswith(bytes((0xEF, 0xBB, 0xBF))):
        raise ConfigError('UTF-8 BOM is not allowed')

    try:
        root_value = parse_json_text(raw.decode('utf-8'))
    except (UnicodeDecodeError, ValueError, RecursionError) as exc:
        raise ConfigError(f'invalid workflow JSON: {exc}') from exc

    root = require_keys(root_value, {'version', 'steps'}, 'root')
    if type(root['version']) is not int or root['version'] != 1:
        raise ConfigError('version must be integer 1')

    raw_steps = root['steps']
    if type(raw_steps) is not list:
        raise ConfigError('steps must be an array')
    if not 0 < len(raw_steps) <= MAX_STEPS:
        raise ConfigError(f'steps must contain 1 to {MAX_STEPS} items')

    steps: list[Step] = []
    used_ids: set[str] = set()

    for index, raw_step in enumerate(raw_steps):
        label = f'steps[{index}]'
        step = require_keys(
            raw_step,
            {'id', 'argv', 'timeout_seconds'},
            label,
        )

        step_id = step['id']
        if type(step_id) is not str or ID_PATTERN.fullmatch(step_id) is None:
            raise ConfigError(f'{label}.id has an invalid format')
        if step_id in used_ids:
            raise ConfigError(f'duplicate step id: {step_id}')
        used_ids.add(step_id)

        raw_argv = step['argv']
        if type(raw_argv) is not list or not 1 <= len(raw_argv) <= 32:
            raise ConfigError(f'{label}.argv must contain 1 to 32 items')

        argv: list[str] = []
        for arg in raw_argv:
            if type(arg) is not str or not arg or len(arg) > 4096:
                raise ConfigError(f'{label}.argv contains an invalid item')
            if any(
                ord(char) < 32
                or ord(char) == 127
                or 0xD800 <= ord(char) <= 0xDFFF
                for char in arg
            ):
                raise ConfigError(
                    f'{label}.argv contains a control character or surrogate'
                )
            argv.append(arg)

        timeout = step['timeout_seconds']
        if type(timeout) is not int or not 1 <= timeout <= 3600:
            raise ConfigError(
                f'{label}.timeout_seconds must be an integer from 1 to 3600'
            )

        steps.append(Step(step_id, tuple(argv), timeout))

    return tuple(steps), hashlib.sha256(raw).hexdigest()


def empty_stream_meta() -> dict[str, object]:
    return {
        'bytes': 0,
        'sha256': hashlib.sha256(b'').hexdigest(),
    }


def read_capture(handle: BinaryIO) -> tuple[bytes | None, dict[str, object]]:
    handle.flush()
    size = os.fstat(handle.fileno()).st_size
    if size > MAX_CAPTURE_BYTES:
        return None, {'bytes': size, 'sha256': None}

    handle.seek(0)
    data = handle.read()
    return data, {
        'bytes': size,
        'sha256': hashlib.sha256(data).hexdigest(),
    }


def stop_process_group(process: subprocess.Popen[bytes]) -> None:
    try:
        os.killpg(process.pid, signal.SIGKILL)
    except ProcessLookupError:
        pass
    except OSError:
        if process.poll() is None:
            process.kill()

    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        process.kill()
        process.wait()


def parse_child_result(data: bytes) -> dict[str, object]:
    if not data.endswith(b'\n'):
        raise ValueError('stdout must end with one LF')

    body = data[:-1]
    if not body or b'\n' in body or b'\r' in body:
        raise ValueError('stdout must contain exactly one line')

    try:
        value = parse_json_text(body.decode('utf-8'))
    except (UnicodeDecodeError, ValueError, RecursionError) as exc:
        raise ValueError(f'stdout is not strict JSON: {exc}') from exc

    if type(value) is not dict:
        raise ValueError('stdout JSON must be an object')
    return value


def run_step(step: Step, cwd: Path) -> dict[str, object]:
    started_ns = time.monotonic_ns()
    empty_meta = empty_stream_meta()
    process: subprocess.Popen[bytes] | None = None

    try:
        with tempfile.TemporaryFile() as stdout_handle, tempfile.TemporaryFile() as stderr_handle:
            argv = [
                sys.executable if item == '{python}' else item
                for item in step.argv
            ]
            process = subprocess.Popen(
                argv,
                cwd=str(cwd),
                stdin=subprocess.DEVNULL,
                stdout=stdout_handle,
                stderr=stderr_handle,
                shell=False,
                start_new_session=True,
            )

            try:
                process.wait(timeout=step.timeout_seconds)
            except subprocess.TimeoutExpired:
                stop_process_group(process)
                _, stdout_meta = read_capture(stdout_handle)
                _, stderr_meta = read_capture(stderr_handle)
                raise StepFailure(
                    5,
                    'timeout',
                    f'timeout after {step.timeout_seconds} seconds',
                    process.returncode,
                    elapsed_ms(started_ns),
                    stdout_meta,
                    stderr_meta,
                )
            except KeyboardInterrupt:
                stop_process_group(process)
                _, stdout_meta = read_capture(stdout_handle)
                _, stderr_meta = read_capture(stderr_handle)
                raise StepFailure(
                    130,
                    'interrupted',
                    'execution was interrupted',
                    process.returncode,
                    elapsed_ms(started_ns),
                    stdout_meta,
                    stderr_meta,
                )

            stdout_data, stdout_meta = read_capture(stdout_handle)
            stderr_data, stderr_meta = read_capture(stderr_handle)
            duration_ms = elapsed_ms(started_ns)

            if process.returncode != 0:
                raise StepFailure(
                    3,
                    'child_failed',
                    f'child exited with status {process.returncode}',
                    process.returncode,
                    duration_ms,
                    stdout_meta,
                    stderr_meta,
                )

            if stdout_data is None or stderr_data is None:
                raise StepFailure(
                    4,
                    'output_too_large',
                    f'each output stream must be at most {MAX_CAPTURE_BYTES} bytes',
                    process.returncode,
                    duration_ms,
                    stdout_meta,
                    stderr_meta,
                )

            if stderr_meta['bytes'] != 0:
                raise StepFailure(
                    4,
                    'unexpected_stderr',
                    'successful child wrote to stderr',
                    process.returncode,
                    duration_ms,
                    stdout_meta,
                    stderr_meta,
                )

            try:
                result = parse_child_result(stdout_data)
            except ValueError as exc:
                raise StepFailure(
                    4,
                    'invalid_stdout',
                    str(exc),
                    process.returncode,
                    duration_ms,
                    stdout_meta,
                    stderr_meta,
                ) from exc

            return {
                'id': step.id,
                'status': 'passed',
                'returncode': process.returncode,
                'duration_ms': duration_ms,
                'stdout': stdout_meta,
                'stderr': stderr_meta,
                'result': result,
            }

    except StepFailure:
        raise
    except OSError as exc:
        if process is not None and process.poll() is None:
            stop_process_group(process)
        raise StepFailure(
            6,
            'launch_or_io_error',
            f'{type(exc).__name__}: {exc}',
            None if process is None else process.returncode,
            elapsed_ms(started_ns),
            empty_meta,
            empty_meta,
        ) from exc


def write_report(run_dir: Path, report: dict[str, object]) -> Path:
    temporary = run_dir / '.report.json.tmp'
    final = run_dir / 'report.json'
    payload = (
        json.dumps(
            report,
            ensure_ascii=True,
            allow_nan=False,
            indent=2,
            sort_keys=True,
        ).encode('utf-8')
        + b'\n'
    )

    with temporary.open('xb') as handle:
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())
    os.replace(temporary, final)
    return final


def main(argv: list[str] | None = None) -> int:
    parser = CliParser()
    parser.add_argument('workflow')
    parser.add_argument('run_dir')

    try:
        args = parser.parse_args(argv)
    except ConfigError as exc:
        emit_error('argument_error', str(exc))
        return 2

    try:
        workflow_path = Path(args.workflow).resolve(strict=True)
        if not workflow_path.is_file():
            raise ConfigError('workflow path is not a regular file')
        steps, config_sha256 = load_workflow(workflow_path)
    except (OSError, RuntimeError, ConfigError) as exc:
        emit_error('config_error', str(exc))
        return 2

    try:
        run_dir_input = Path(args.run_dir)
        os.mkdir(run_dir_input, 0o700)
        run_dir = run_dir_input.resolve(strict=True)
    except (OSError, RuntimeError) as exc:
        emit_error('run_directory_error', str(exc))
        return 6

    started_at = utc_now()
    results: list[dict[str, object]] = []
    status = 'passed'
    failed_step: str | None = None
    uncompleted: list[str] = []
    exit_code = 0
    next_index = 0

    try:
        while next_index < len(steps):
            step = steps[next_index]
            try:
                result = run_step(step, workflow_path.parent)
            except StepFailure as exc:
                results.append(exc.to_record(step.id))
                status = 'interrupted' if exc.exit_code == 130 else 'failed'
                failed_step = step.id
                uncompleted = [item.id for item in steps[next_index + 1:]]
                exit_code = exc.exit_code
                break

            results.append(result)
            next_index += 1
    except KeyboardInterrupt:
        status = 'interrupted'
        uncompleted = [item.id for item in steps[next_index:]]
        exit_code = 130

    report: dict[str, object] = {
        'schema_version': 1,
        'status': status,
        'config_sha256': config_sha256,
        'started_at': started_at,
        'finished_at': utc_now(),
        'failed_step': failed_step,
        'uncompleted_step_ids': uncompleted,
        'steps': results,
    }

    try:
        report_path = write_report(run_dir, report)
    except (OSError, TypeError, ValueError, RecursionError) as exc:
        emit_error('report_write_error', str(exc))
        return 6

    print(compact({
        'status': status,
        'report': str(report_path),
        'failed_step': failed_step,
    }))
    return exit_code


def entry() -> int:
    try:
        return main()
    except KeyboardInterrupt:
        emit_error('interrupted', 'execution was interrupted')
        return 130


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

4.正常系を実行する

run-okがまだ存在しないことを確認してから、3ファイルを置いたディレクトリで実行します。

python3 workflow_runner.py workflow.json run-ok
runner_status=$?
printf 'exit=%s\n' "$runner_status"
python3 -m json.tool run-ok/report.json

すべての子CLIが契約を満たすため、実行器の終了コードは0になります。レポートのstatusはpassed、3件のステップもすべてpassedとなり、uncompleted_step_idsは空です。

同じrun-okを指定して再実行すると、既存ディレクトリを上書きせず終了コード6で停止します。既存のファイルやシンボリックリンクと同じ名前も使えません。実行ごとに新しいディレクトリ名を割り当ててください。

5.途中で失敗した場合を確認する

次の内容をworkflow_fail.jsonとして保存します。2番目のステップだけに--failを追加しています。

{
  "version": 1,
  "steps": [
    {
      "id": "input_check",
      "argv": ["{python}", "sample_gate.py", "input_check"],
      "timeout_seconds": 10
    },
    {
      "id": "candidate_check",
      "argv": ["{python}", "sample_gate.py", "candidate_check", "--fail"],
      "timeout_seconds": 10
    },
    {
      "id": "publish_check",
      "argv": ["{python}", "sample_gate.py", "publish_check"],
      "timeout_seconds": 10
    }
  ]
}

run-failが存在しない状態で実行します。

python3 workflow_runner.py workflow_fail.json run-fail
runner_status=$?
printf 'exit=%s\n' "$runner_status"
python3 -m json.tool run-fail/report.json

実行器の終了コードは3になります。レポートではinput_checkが成功し、candidate_checkのreturncodeが7になります。3番目のpublish_checkは実行されず、uncompleted_step_idsへ記録されます。

実行器は、失敗した子CLIの標準出力や標準エラー出力をレポートへ複製しません。上限内ならバイト数とSHA-256を残し、65,536バイトを超えたストリームはバイト数とnullを記録します。入力値や個人情報をエラーへ出すCLIを誤って接続した場合に、レポートへ同じ情報を増やさないためです。

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

終了コード 意味 対応
0 全ステップが成功 レポートを確認し、次の工程へ進む
2 引数または設定JSONが不正 設定の項目、型、重複キーを直す
3 子CLIが非ゼロで終了 該当CLIの入力と個別レポートを確認する
4 子CLIが0で終了したが、成功時の出力契約に違反 標準出力、標準エラー出力、JSON形式を直す
5 タイムアウト 処理量、外部待ち、制限時間を確認する
6 起動、ディレクトリ、レポート書き込みの失敗 パス、権限、空き容量を確認する
130 Ctrl+Cによる中断 途中生成物を確認してから新しい実行を開始する

子CLI固有の終了コードは、そのまま実行器の終了コードにはしません。たとえば子CLIの終了コード7はレポートへ保存し、実行器は一律に3を返します。これにより、実行器を呼び出す側は共通の分類だけを扱えます。

実務へ置き換えるときの注意点

  • 成功時の標準出力には、件数、判定、レポートパスなどの業務メタデータだけを出す
  • 大量ログを出す処理は接続しない。終了コード0でも、各出力が65,536バイトを超えれば契約違反として停止する
  • 出力は一時ファイルまたは新しいパスへ作り、検査完了前のファイルを公開名へ切り替えない
  • タイムアウト時はプロセスグループへSIGKILLを送るため、子CLIの終了処理は実行されない
  • バックグラウンド化や別セッション化を行うCLIは対象にしない
  • 設定のargvへAPIキーやパスワードを書かない。引数は同じ端末の別プロセスから見える場合がある
  • 子CLIは親プロセスの環境変数を継承する。実行用アカウントへ必要以上の秘密情報を持たせない

標準出力と標準エラー出力は、一時ファイルへ受けてからサイズを確認します。このためメモリへ無制限に読み込むことはありませんが、終了またはタイムアウトまでは一時領域を消費します。大量のログを生成するコマンドを制御する用途には、ログ量を実行中に制限する別の仕組みが必要です。

config_sha256は、どの設定バイト列を読み込んだ実行かを照合するための値です。署名ではないため、設定とレポートの両方を書き換えられる利用者に対する改ざん証明にはなりません。

最初は二つの処理から接続する

実務へ導入するときは、検査と変換など、依存関係が明確な二つのCLIから始めます。それぞれを単独で実行し、終了コード、標準出力、標準エラー出力を確認してから設定へ追加します。

この実行器が保証するのは、登録した順序で一つずつ起動し、失敗後のステップを実行しないことです。各処理の正しさ、完了済み変更の取り消し、外部サービス側の結果までは保証しません。その境界を分けることで、小さなPython処理を一つの業務手順へまとめられます。

第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の場合も、アプリケーションを停止したまま実際の置換直前に再検査し、承認済みのロールバック手順と組み合わせる必要があります。

第170回 検証済みSQLiteを取り違えずに切り替える:ハッシュ固定と原子的置換

第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())
PY
  • CANDIDATE_SHAと一致する場合、候補の内容が稼働用パスにあります。レポートと後段検査の失敗理由を確認します。
  • LIVE_SHAと一致する場合、切り替え前の内容です。
  • どちらとも一致しない場合は、起動や再実行をせず、メインファイルと副ファイルを保全して調査します。

アプリ起動後に退避コピーを戻すと、新しい更新を失う可能性があります。この退避コピーは、自動ロールバックを許可するものではありません。

適用上の制約

  • プロセス間ロックは実装していません。コマンドを並行実行せず、対象ディレクトリをほかの処理から変更しない運用が必要です。
  • 単一のSQLiteメインファイルだけを対象とします。ATTACHしたデータベースや外部ファイルは切り替えません。
  • WAL形式は、副ファイルが存在しない場合も対象外です。
  • ネットワークファイルシステム、Windows、コンテナ間で共有するボリュームにはそのまま適用しません。
  • 退避コピーとレポートにはfsync()を使いますが、ディレクトリエントリ、ストレージ装置、突然の電源断まで含む耐久性は保証しません。
  • コードは候補の構造整合性とバージョンを再確認しますが、移行固有の書き込み試験や検索結果の突合は再実行しません。記録済みの候補ハッシュによって、事前検査したファイルと切り替えるファイルを結び付けます。

status=passedが示すのは、指定したデータベースファイルの切り替えと再検査が完了したことです。アプリケーション全体の正常稼働は、別の起動手順とヘルスチェックで確認します。

第169回 SQLite移行前後の検索結果を突合する:Pythonで作る読み取り回帰テスト

書き込み試験に合格しても、既存の一覧や集計が移行前と同じ結果を返すとは限りません。

今回は、同じスナップショットを起点とする移行前後のSQLiteへ、既存業務を代表する3本のSELECTを実行します。列名、行順、値、SQLiteから返された型を突き合わせ、差分があれば切り替えを止めます。

稼働中のデータベースや、取得時点の異なるファイルは比較対象にしません。

3種類の試験を分ける

試験 確認できること 確認できないこと
構造検査 列、制約、インデックス、スキーマバージョン 既存の検索が同じ結果を返すか
書き込み試験 登録、更新、拒否、ロールバック 一覧や集計の互換性
読み取り回帰テスト 指定した検索の列名、行順、値、型 指定していない検索や、結果自体の業務上の正しさ

新しく追加した列の機能確認など、移行後だけに存在する動作は、移行前との比較ではなく別の期待値テストで確認します。

今回の実行契約

項目 仕様
動作環境 Python 3.10以上、外部ライブラリ不要
入力 同じスナップショットを起点とする、移行前後の独立したSQLiteファイル
入力の状態 書き込みが終了し、隣に-wal、-shm、-journalがないこと
接続 URIのmode=roとPRAGMA query_only=ONを使用する
比較 列名、行数、行順、値、値の型を完全一致で比較する
生成物 新しいJSONレポート。既存ファイルは上書きしない
レポート内容 件数、列名、結果のSHA-256、最初の差分の種類と位置。セル値は記録しない
対象外 性能測定、意図したデータ変換の承認、稼働中DBの比較、未指定クエリの保証

ファイルを読み取り専用で開く指定はmode=roです。query_onlyも設定しますが、これだけでデータベースファイルが完全な読み取り専用になるわけではありません。

複数行を返すクエリでは、一意になる並び順を指定します。同順位が残る場合は主キーまでORDER BYへ追加します。

1.再現用のSQLiteを作る

新しい空のディレクトリを用意します。コマンド例はPOSIX系シェル向けです。

mkdir sqlite-read-regression-demo
cd sqlite-read-regression-demo

次のコードをmake_fixture.pyとして保存します。移行前のbaseline.sqliteを作り、そのファイルを複製してからpriority列を追加したcandidate.sqliteを作ります。データは動作確認用です。

import json
import shutil
import sqlite3
from pathlib import Path

BASELINE = Path('baseline.sqlite')
CANDIDATE = Path('candidate.sqlite')
OUTPUTS = (BASELINE, CANDIDATE)

SCHEMA_V1 = '''
CREATE TABLE review_items (
    batch_id TEXT NOT NULL,
    record_id TEXT NOT NULL,
    proposed_category TEXT NOT NULL,
    status TEXT NOT NULL CHECK (
        status IN ('pending', 'accepted', 'corrected', 'rejected')
    ),
    confirmed_category TEXT,
    PRIMARY KEY (batch_id, record_id)
);
'''

ROWS = (
    ('B001', 'R001', '配送', 'pending', None),
    ('B001', 'R002', '請求', 'accepted', '請求'),
    ('B001', 'R003', '配送', 'corrected', 'その他'),
    ('B002', 'R004', '請求', 'rejected', None),
)


def create_baseline(path):
    with path.open('xb'):
        pass

    connection = sqlite3.connect(path)
    try:
        connection.executescript(SCHEMA_V1)
        connection.executemany(
            '''
            INSERT INTO review_items (
                batch_id,
                record_id,
                proposed_category,
                status,
                confirmed_category
            ) VALUES (?, ?, ?, ?, ?)
            ''',
            ROWS,
        )
        connection.execute('PRAGMA user_version = 1')
        connection.commit()
    except BaseException:
        connection.rollback()
        raise
    finally:
        connection.close()


def copy_database(source, destination):
    with source.open('rb') as source_handle:
        with destination.open('xb') as destination_handle:
            shutil.copyfileobj(source_handle, destination_handle)


def migrate_candidate(path):
    connection = sqlite3.connect(path)
    try:
        connection.execute(
            '''
            ALTER TABLE review_items
            ADD COLUMN priority INTEGER NOT NULL DEFAULT 0
                CHECK (priority BETWEEN 0 AND 3)
            '''
        )
        connection.execute('PRAGMA user_version = 2')
        connection.commit()
    except BaseException:
        connection.rollback()
        raise
    finally:
        connection.close()


def main():
    if any(path.exists() for path in OUTPUTS):
        raise SystemExit('run this fixture in a new empty directory')

    try:
        create_baseline(BASELINE)
        copy_database(BASELINE, CANDIDATE)
        migrate_candidate(CANDIDATE)
    except BaseException:
        for path in OUTPUTS:
            path.unlink(missing_ok=True)
        raise

    print(json.dumps(
        {'files': [str(path) for path in OUTPUTS], 'status': 'created'},
        ensure_ascii=False,
        separators=(',', ':'),
        sort_keys=True,
    ))


if __name__ == '__main__':
    main()

2.読み取り回帰テストを実装する

次のコードをsqlite_read_regression.pyとして保存します。実行するSQLはCASESへ固定し、コマンドラインから任意のSQLを渡せない形にします。

import argparse
import hashlib
import json
import os
import sqlite3
import sys
from pathlib import Path

MAX_ROWS = 10_000
MAX_COLUMNS = 100
MAX_RESULT_BYTES = 20 * 1024 * 1024
PROGRESS_INTERVAL = 1_000
MAX_PROGRESS_CALLS = 20_000

CASES = (
    {
        'name': 'pending_items',
        'sql': '''SELECT batch_id, record_id, proposed_category, status
FROM review_items
WHERE status = ?
ORDER BY batch_id, record_id''',
        'params': ('pending',),
    },
    {
        'name': 'status_counts',
        'sql': '''SELECT status, COUNT(*) AS item_count
FROM review_items
GROUP BY status
ORDER BY status''',
        'params': (),
    },
    {
        'name': 'final_categories',
        'sql': '''SELECT
    batch_id,
    record_id,
    CASE
        WHEN status = 'corrected' THEN confirmed_category
        ELSE proposed_category
    END AS final_category
FROM review_items
WHERE status IN (?, ?)
ORDER BY batch_id, record_id''',
        'params': ('accepted', 'corrected'),
    },
)


class InputError(Exception):
    pass


class CaseRunError(Exception):
    pass


def compact_bytes(value):
    return json.dumps(
        value,
        ensure_ascii=False,
        separators=(',', ':'),
        sort_keys=True,
        allow_nan=False,
    ).encode('utf-8')


def emit(stream, value):
    print(json.dumps(
        value,
        ensure_ascii=False,
        separators=(',', ':'),
        sort_keys=True,
        allow_nan=False,
    ), file=stream, flush=True)


def sha256_file(path):
    digest = hashlib.sha256()
    with path.open('rb') as handle:
        while True:
            chunk = handle.read(1024 * 1024)
            if not chunk:
                break
            digest.update(chunk)
    return digest.hexdigest()


def find_sidecars(entries):
    found = []
    for label, path in entries:
        for suffix in ('-wal', '-shm', '-journal'):
            sidecar = path.with_name(path.name + suffix)
            if sidecar.exists():
                found.append((label, sidecar))
    return found


def validate_inputs(baseline_path, candidate_path, report_path):
    for label, path in (
        ('baseline', baseline_path),
        ('candidate', candidate_path),
    ):
        if not path.is_file():
            raise InputError(f'{label} is not a file: {path}')

    if baseline_path.samefile(candidate_path):
        raise InputError('baseline and candidate must be different files')
    if report_path.exists():
        raise InputError(f'report already exists: {report_path}')
    if not report_path.parent.is_dir():
        raise InputError(f'report directory does not exist: {report_path.parent}')

    names = set()
    for case in CASES:
        name = case['name']
        if name in names:
            raise InputError(f'duplicate case name: {name}')
        names.add(name)

        words = case['sql'].split(None, 1)
        if not words or words[0].upper() != 'SELECT':
            raise InputError(f'case must start with SELECT: {name}')


def open_readonly(path):
    uri = path.resolve().as_uri() + '?mode=ro'
    connection = sqlite3.connect(
        uri,
        uri=True,
        isolation_level=None,
        timeout=5.0,
    )
    try:
        connection.execute('PRAGMA query_only = ON')
        connection.execute('BEGIN')
        connection.execute('SELECT COUNT(*) FROM sqlite_master').fetchone()
        return connection
    except BaseException:
        connection.close()
        raise


def normalize_value(value):
    if value is None:
        return ['null', None]
    if type(value) is int:
        return ['integer', str(value)]
    if type(value) is float:
        return ['real', value.hex()]
    if type(value) is str:
        return ['text', value]
    if isinstance(value, (bytes, bytearray, memoryview)):
        return ['blob', bytes(value).hex()]
    raise CaseRunError(
        f'unsupported SQLite result type: {type(value).__name__}'
    )


def add_digest_piece(digest, piece):
    digest.update(len(piece).to_bytes(8, 'big'))
    digest.update(piece)


def read_result(connection, case):
    calls = 0
    cursor = None

    def progress():
        nonlocal calls
        calls += 1
        return 1 if calls > MAX_PROGRESS_CALLS else 0

    connection.set_progress_handler(progress, PROGRESS_INTERVAL)
    try:
        cursor = connection.execute(case['sql'], case['params'])
        if cursor.description is None:
            raise CaseRunError('query did not return a result set')

        columns = [description[0] for description in cursor.description]
        if len(columns) > MAX_COLUMNS:
            raise CaseRunError('column limit exceeded')

        digest = hashlib.sha256()
        header_piece = compact_bytes(['columns', columns])
        total_bytes = len(header_piece)
        if total_bytes > MAX_RESULT_BYTES:
            raise CaseRunError('result byte limit exceeded')
        add_digest_piece(digest, header_piece)

        rows = []
        while True:
            row = cursor.fetchone()
            if row is None:
                break
            if len(rows) >= MAX_ROWS:
                raise CaseRunError('row limit exceeded')

            normalized = [normalize_value(value) for value in row]
            row_piece = compact_bytes(normalized)
            total_bytes += len(row_piece)
            if total_bytes > MAX_RESULT_BYTES:
                raise CaseRunError('result byte limit exceeded')

            add_digest_piece(digest, row_piece)
            rows.append(normalized)

        return {
            'columns': columns,
            'rows': rows,
            'row_count': len(rows),
            'sha256': digest.hexdigest(),
        }
    except sqlite3.OperationalError as exc:
        if calls > MAX_PROGRESS_CALLS:
            raise CaseRunError('query operation budget exceeded') from exc
        raise
    finally:
        connection.set_progress_handler(None, 0)
        if cursor is not None:
            cursor.close()


def definition_hash(case):
    payload = {
        'params': list(case['params']),
        'sql': case['sql'],
    }
    return hashlib.sha256(compact_bytes(payload)).hexdigest()


def result_summary(result):
    return {
        'columns': result['columns'],
        'row_count': result['row_count'],
        'sha256': result['sha256'],
    }


def compare_case(baseline_connection, candidate_connection, case):
    payload = {
        'definition_sha256': definition_hash(case),
        'name': case['name'],
    }
    results = {}
    errors = {}

    for label, connection in (
        ('baseline', baseline_connection),
        ('candidate', candidate_connection),
    ):
        try:
            results[label] = read_result(connection, case)
        except (sqlite3.Error, CaseRunError, UnicodeError) as exc:
            errors[label] = {
                'message': str(exc),
                'type': type(exc).__name__,
            }

    if errors:
        for label, result in results.items():
            payload[label] = result_summary(result)
        payload['errors'] = errors
        payload['status'] = 'error'
        return payload

    baseline = results['baseline']
    candidate = results['candidate']
    payload['baseline'] = result_summary(baseline)
    payload['candidate'] = result_summary(candidate)

    first_difference = None
    if baseline['columns'] != candidate['columns']:
        common_columns = min(
            len(baseline['columns']),
            len(candidate['columns']),
        )
        column_index = common_columns + 1
        for index in range(common_columns):
            if baseline['columns'][index] != candidate['columns'][index]:
                column_index = index + 1
                break
        first_difference = {
            'column_index': column_index,
            'kind': 'columns',
        }
    else:
        common_rows = min(baseline['row_count'], candidate['row_count'])
        for row_index in range(common_rows):
            baseline_row = baseline['rows'][row_index]
            candidate_row = candidate['rows'][row_index]
            if baseline_row == candidate_row:
                continue

            for cell_index, (baseline_cell, candidate_cell) in enumerate(
                zip(baseline_row, candidate_row)
            ):
                if baseline_cell != candidate_cell:
                    difference_kind = (
                        'type'
                        if baseline_cell[0] != candidate_cell[0]
                        else 'value'
                    )
                    first_difference = {
                        'column': baseline['columns'][cell_index],
                        'column_index': cell_index + 1,
                        'kind': difference_kind,
                        'row_number': row_index + 1,
                    }
                    break
            break

        if first_difference is None and (
            baseline['row_count'] != candidate['row_count']
        ):
            first_difference = {
                'kind': 'row_count',
                'row_number': common_rows + 1,
            }

    if first_difference is None:
        payload['status'] = 'passed'
    else:
        payload['first_difference'] = first_difference
        payload['status'] = 'failed'
    return payload


def write_report(report_path, document):
    payload = json.dumps(
        document,
        ensure_ascii=False,
        indent=2,
        sort_keys=True,
        allow_nan=False,
    )
    created = False
    try:
        with report_path.open('x', encoding='utf-8', newline='') as handle:
            created = True
            handle.write(payload)
            handle.write(chr(10))
            handle.flush()
            os.fsync(handle.fileno())
    except FileExistsError as exc:
        raise InputError(f'report already exists: {report_path}') from exc
    except BaseException:
        if created:
            try:
                report_path.unlink()
            except OSError:
                pass
        raise


def run(args):
    baseline_path = args.baseline
    candidate_path = args.candidate
    report_path = args.report
    validate_inputs(baseline_path, candidate_path, report_path)

    entries = (
        ('baseline', baseline_path),
        ('candidate', candidate_path),
    )
    initial_sidecars = find_sidecars(entries)
    if initial_sidecars:
        details = ', '.join(
            f'{label}:{path.name}' for label, path in initial_sidecars
        )
        raise InputError(f'SQLite sidecar found: {details}')

    initial_hashes = {
        'baseline': sha256_file(baseline_path),
        'candidate': sha256_file(candidate_path),
    }

    baseline_connection = None
    candidate_connection = None
    try:
        baseline_connection = open_readonly(baseline_path)
        candidate_connection = open_readonly(candidate_path)
        baseline_version = baseline_connection.execute(
            'PRAGMA user_version'
        ).fetchone()[0]
        candidate_version = candidate_connection.execute(
            'PRAGMA user_version'
        ).fetchone()[0]
        case_reports = [
            compare_case(baseline_connection, candidate_connection, case)
            for case in CASES
        ]
    finally:
        for connection in (candidate_connection, baseline_connection):
            if connection is not None:
                connection.close()

    final_hashes = {
        'baseline': sha256_file(baseline_path),
        'candidate': sha256_file(candidate_path),
    }
    post_sidecars = find_sidecars(entries)

    stability_problems = []
    for label in ('baseline', 'candidate'):
        if initial_hashes[label] != final_hashes[label]:
            stability_problems.append(f'{label}_changed_during_test')
    for label, path in post_sidecars:
        stability_problems.append(
            f'{label}_sidecar_appeared:{path.name}'
        )

    if stability_problems:
        overall_status = 'error'
        exit_code = 2
    elif all(case['status'] == 'passed' for case in case_reports):
        overall_status = 'passed'
        exit_code = 0
    else:
        overall_status = 'failed'
        exit_code = 1

    document = {
        'case_count': len(case_reports),
        'cases': case_reports,
        'input_stability': {
            'problems': stability_problems,
            'status': 'passed' if not stability_problems else 'failed',
        },
        'inputs': {
            'baseline': {
                'file': baseline_path.name,
                'sha256': initial_hashes['baseline'],
                'user_version': baseline_version,
            },
            'candidate': {
                'file': candidate_path.name,
                'sha256': initial_hashes['candidate'],
                'user_version': candidate_version,
            },
        },
        'status': overall_status,
        'version': 1,
    }
    write_report(report_path, document)
    emit(sys.stdout, {
        'case_count': len(case_reports),
        'report': str(report_path),
        'status': overall_status,
    })
    return exit_code


def main(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument('baseline', type=Path)
    parser.add_argument('candidate', type=Path)
    parser.add_argument('report', type=Path)
    args = parser.parse_args(argv)

    try:
        return run(args)
    except InputError as exc:
        emit(sys.stderr, {'error': str(exc), 'status': 'input_error'})
        return 2
    except KeyboardInterrupt:
        emit(sys.stderr, {'error': 'interrupted', 'status': 'interrupted'})
        return 130
    except (OSError, sqlite3.Error) as exc:
        emit(sys.stderr, {
            'error': str(exc),
            'error_type': type(exc).__name__,
            'status': 'execution_error',
        })
        return 3
    except Exception as exc:
        emit(sys.stderr, {
            'error': str(exc),
            'error_type': type(exc).__name__,
            'status': 'unexpected_error',
        })
        return 3


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

3.正常系を実行する

python make_fixture.py
python sqlite_read_regression.py baseline.sqlite candidate.sqlite report.json
echo $?

標準出力と終了コードは次のようになります。

{"files":["baseline.sqlite","candidate.sqlite"],"status":"created"}
{"case_count":3,"report":"report.json","status":"passed"}
0

レポートは標準ライブラリで整形できます。

python -m json.tool report.json
項目 意味
definition_sha256 SQLとバインド値をまとめたテスト定義の識別値
columns SQLiteが返した列名と順序
row_count 結果行数
sha256 型付きで正規化した列名と全行の結果ハッシュ
first_difference 最初に見つかった差分の種類と、1から始まる列番号または行番号
input_stability 試験中に入力ファイルやサイドカーファイルが変化しなかったか

first_difference.kindは、列構成の差ならcolumns、行数の差ならrow_count、セルの型の差ならtype、同じ型の値の差ならvalueです。列構成の差には最初に異なるcolumn_index、セルの差にはrow_number、column_index、列名を記録します。行数の差では、片方だけに存在する最初の行番号をrow_numberへ記録します。

値は内部でinteger、real、text、blob、nullのタグ付き表現へ変換します。Pythonでは1 == 1.0が真になりますが、このテストでは整数と実数を別の型として検出します。

4.差分を検出する

次の操作は、ここで作ったcandidate.sqliteだけに実行してください。

make_difference.pyとして保存します。

import sqlite3

connection = sqlite3.connect('candidate.sqlite')
try:
    cursor = connection.execute(
        '''
        UPDATE review_items
        SET status = ?, confirmed_category = ?
        WHERE batch_id = ? AND record_id = ?
        ''',
        ('accepted', '配送', 'B001', 'R001'),
    )
    if cursor.rowcount != 1:
        raise RuntimeError('expected exactly one updated row')
    connection.commit()
except BaseException:
    connection.rollback()
    raise
finally:
    connection.close()

report.jsonは上書きできないため、新しい出力名を指定します。

python make_difference.py
python sqlite_read_regression.py baseline.sqlite candidate.sqlite report-diff.json
echo $?

検索結果が一致しないため、終了コードは1になります。

{"case_count":3,"report":"report-diff.json","status":"failed"}
1

report-diff.jsonには、各失敗ケースで最初に見つかった差分の種類と位置が記録されます。セル値は複製しません。ただし、列名、件数、ハッシュも業務データから派生した情報です。レポートには元データに準じたアクセス制御を適用します。

終了コードを後続処理へ渡す

終了コード 意味 判断
0 全クエリが一致し、入力も変化しなかった ほかの切り替え条件の確認へ進む
1 結果差分またはクエリ実行エラー 切り替えを止め、SQLと移行内容を確認する
2 入力契約違反、既存レポート、サイドカー、入力内容の変化 隔離コピーと実行条件を確認する
3 ファイル操作、SQLite接続、予期しない実行エラー 実行環境を確認する
130 Ctrl+Cによる中断 未完了として扱う

argparseが引数不足を検出した場合も終了コードは2です。この場合は通常の使用方法が表示され、1行JSONにはなりません。

実データへ合わせるときの注意点

  • 稼働中DBを直接指定しない:SQLiteのバックアップAPIなどで閉じたスナップショットを作り、移行候補はそのスナップショットから作成します。
  • 重要な検索を選ぶ:未処理一覧、確定結果、請求対象など、切り替え後に使う検索を少数から登録します。
  • 並び順を固定する:複数行を比較する場合は、一意キーまで含めたORDER BYを指定します。ORDER BYがない結果や、同順位が残る結果の順序は保証されません。
  • 意図した差分を混ぜない:移行で値を変える仕様なら、完全一致の条件を緩めず、変換後の期待値を確認する別テストに分けます。
  • 同じ誤りは検出できない:移行前後が同じ誤った値を返せば合格します。業務上の正解を確認するテストの代わりにはなりません。
  • 大規模な抽出には使わない:1ケースあたり1万行、100列、正規化したJSON断片の合計20MiBで停止します。進捗ハンドラーによる上限は、処理時間の上限ではありません。
  • ハッシュを改ざん証明にしない:SHA-256は対象や結果の取り違え検出には使えますが、データベースとレポートの両方を変更できる利用者に対する改ざん証明にはなりません。

最初は、切り替え後に必ず使う検索を1本だけCASESへ移します。隔離コピーで合格条件を確認してから、一覧、集計、確定データの順に増やします。

一次資料

第168回 SQLite移行後の書き込みを本番前に試す:Pythonで作る破棄可能なスモークテスト

第167回では、SQLiteのスキーマ変更を本番データベースへ直接適用せず、コピー上で実行しました。列定義や整合性を確認しても、移行後に予定している登録や状態更新が正しく動くとは限りません。

今回は、移行済みの候補データベースを読み取り専用で開き、OSの一時領域へ複製します。そのコピー上で、正常な登録と更新、制約違反の拒否、ロールバック後の無残留を確認します。元の候補データベースには書き込みません。

サンプルは、第167回の考え方を引き継いだ独立した動作確認用スキーマです。実際に使うときは、列名、状態遷移、外部キー、試験値を業務スキーマに合わせて変更してください。

構造検査と書き込み試験を分ける

構造検査では、必要な列やインデックスの存在を確認します。書き込み試験では、アプリケーションが使用するSQLと制約の組み合わせを確認します。

確認対象 今回の判定
正常な登録 priorityを省略した登録が成功し、既定値0になる
正常な更新 優先度3への変更と、確認済み状態への遷移が成功する
範囲制約 優先度4への変更がsqlite3.IntegrityErrorになる
一意性 同じバッチIDとレコードIDの再登録が拒否される
状態制約 確定カテゴリがないままacceptedへ変更できない
後始末 試験用トランザクションを取り消すと、行数と試験用IDが元に戻る
データベース検査 integrity_check(1)とforeign_key_checkを通過する

今回の実行契約

項目 仕様
動作環境 Python 3.10以上、SQLite 3.37.0以上
入力 書き込みを停止した、単独の移行済みSQLiteファイル
入力の扱い mode=roとimmutable=1を指定して開き、SQLiteのバックアップAPIで一時ファイルへ複製する
入力のサイドカー -wal、-shm、-journalが残っている入力は拒否する
サンプル出力 DB本体またはいずれかのサイドカー名が既に存在する場合は作成を拒否する
書き込み試験の書き込み先 TemporaryDirectory内のコピーだけ
永続的な出力 make-sampleで作る動作確認用DBを除き、なし
結果 ヘルプ表示を除き、標準出力または標準エラー出力へ1行のJSONを返す
対象外 本番切り替え、アプリケーション全体の互換性、同時書き込み、性能試験、旧版への逆移行

-walファイルには、メインファイルへ反映されていない確定済みデータが含まれる場合があります。検査を通すためにサイドカーファイルを手動で削除してはいけません。書き込み元を正常に停止し、必要なチェックポイントや復旧をSQLiteに実行させてから候補ファイルを用意します。

make-sampleも、指定したDB本体だけでなく同名の-wal、-shm、-journalを作成前に検査します。既存のものが一つでもあれば処理を始めません。失敗時に削除するのは、排他的作成で今回生成したDB本体と同じファイルであることをデバイス番号とinode番号で確認できた場合だけです。既存または後から現れたサイドカーは削除しません。

この確認は、出力ディレクトリで別の処理が同じ名前を同時に作成、置換する状況を調停するロックではありません。サンプル作成中に同じ出力名を操作しない、信頼できるディレクトリで実行してください。

一時コピーにも入力データが含まれます。機密データを扱う場合は、一時領域の保存先と権限を確認してください。TemporaryDirectoryによる削除は、強制終了や電源断を含む安全な消去を保証しません。

スモークテストを実装する

次のコードをsmoke_review_db.pyとして保存します。外部ライブラリは使いません。

from __future__ import annotations

import argparse
import hashlib
import json
import os
import secrets
import sqlite3
import stat
import sys
import tempfile
from pathlib import Path
from typing import TextIO


EXPECTED_VERSION = 2
REQUIRED_COLUMNS = frozenset(
    {
        'batch_id',
        'record_id',
        'proposed_category',
        'status',
        'priority',
        'decided_category',
    }
)

SAMPLE_SQL = '''
BEGIN IMMEDIATE;

CREATE TABLE review_items (
    batch_id TEXT NOT NULL CHECK (length(batch_id) > 0),
    record_id TEXT NOT NULL CHECK (length(record_id) > 0),
    proposed_category TEXT NOT NULL
        CHECK (length(proposed_category) > 0),
    status TEXT NOT NULL DEFAULT 'pending'
        CHECK (
            status IN ('pending', 'accepted', 'corrected', 'rejected')
        ),
    priority INTEGER NOT NULL DEFAULT 0
        CHECK (priority BETWEEN 0 AND 3),
    decided_category TEXT,
    PRIMARY KEY (batch_id, record_id),
    CHECK (
        (status = 'pending' AND decided_category IS NULL)
        OR
        (status IN ('accepted', 'corrected')
            AND decided_category IS NOT NULL)
        OR
        (status = 'rejected' AND decided_category IS NULL)
    )
) STRICT;

CREATE INDEX review_items_pending_priority_idx
    ON review_items(priority DESC, batch_id, record_id)
    WHERE status = 'pending';

INSERT INTO review_items (
    batch_id, record_id, proposed_category, status
) VALUES (
    'sample-batch', 'sample-record', 'sample-category', 'pending'
);

PRAGMA user_version = 2;
COMMIT;
'''


class ContractError(Exception):
    pass


def json_line(payload: dict[str, object], stream: TextIO) -> None:
    print(
        json.dumps(payload, ensure_ascii=False, separators=(',', ':')),
        file=stream,
    )


class JsonArgumentParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        json_line(
            {'status': 'error', 'kind': 'argument', 'message': message},
            sys.stderr,
        )
        raise SystemExit(2)


def require_sqlite() -> None:
    if sqlite3.sqlite_version_info < (3, 37, 0):
        raise ContractError(
            f'SQLite 3.37.0以上が必要です: {sqlite3.sqlite_version}'
        )


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 sidecar_paths(path: Path) -> tuple[Path, ...]:
    return tuple(
        Path(f'{path}{suffix}')
        for suffix in ('-wal', '-shm', '-journal')
    )


def existing_paths(paths: tuple[Path, ...]) -> list[Path]:
    return [path for path in paths if os.path.lexists(path)]


def reject_sample_output_paths(path: Path) -> None:
    present = existing_paths((path, *sidecar_paths(path)))
    if present:
        raise ContractError(
            '出力先に既存ファイルがあります: '
            + ','.join(candidate.name for candidate in present)
        )


def file_identity_from_fd(descriptor: int) -> tuple[int, int]:
    information = os.fstat(descriptor)
    return information.st_dev, information.st_ino


def has_file_identity(path: Path, identity: tuple[int, int]) -> bool:
    try:
        information = path.lstat()
    except FileNotFoundError:
        return False
    return (
        stat.S_ISREG(information.st_mode)
        and (information.st_dev, information.st_ino) == identity
    )


def cleanup_created_database(
    path: Path,
    identity: tuple[int, int],
) -> None:
    if not has_file_identity(path, identity):
        return
    try:
        path.unlink()
    except FileNotFoundError:
        pass


def make_sample(path: Path) -> dict[str, object]:
    require_sqlite()
    path = Path(os.path.abspath(path))
    reject_sample_output_paths(path)

    try:
        descriptor = os.open(
            path,
            os.O_WRONLY | os.O_CREAT | os.O_EXCL,
            0o600,
        )
    except FileExistsError as exc:
        raise ContractError(
            f'出力先が既に存在します: {path}'
        ) from exc

    created_identity = file_identity_from_fd(descriptor)
    descriptor_open = True
    try:
        os.close(descriptor)
        descriptor_open = False

        present_sidecars = existing_paths(sidecar_paths(path))
        if present_sidecars:
            raise ContractError(
                '出力先にサイドカーファイルがあります: '
                + ','.join(
                    candidate.name for candidate in present_sidecars
                )
            )

        connection = sqlite3.connect(path)
        try:
            journal = connection.execute(
                'PRAGMA journal_mode = MEMORY'
            ).fetchone()
            if journal is None or str(journal[0]).lower() != 'memory':
                raise ContractError(
                    'サンプルDBをMEMORYジャーナルへ変更できません'
                )

            connection.execute('PRAGMA foreign_keys = ON')
            connection.executescript(SAMPLE_SQL)
            rows = int(
                connection.execute(
                    'SELECT count(*) FROM review_items'
                ).fetchone()[0]
            )
        finally:
            connection.close()

        if not has_file_identity(path, created_identity):
            raise ContractError('作成したDB本体が置き換えられました')

        present_sidecars = existing_paths(sidecar_paths(path))
        if present_sidecars:
            raise ContractError(
                '作成後にサイドカーファイルが見つかりました: '
                + ','.join(
                    candidate.name for candidate in present_sidecars
                )
            )
    except BaseException:
        if descriptor_open:
            try:
                os.close(descriptor)
            except OSError:
                pass
        cleanup_created_database(path, created_identity)
        raise

    return {
        'status': 'created',
        'command': 'make-sample',
        'database': path.name,
        'user_version': EXPECTED_VERSION,
        'rows': rows,
    }


def reject_sidecars(path: Path) -> None:
    present = existing_paths(sidecar_paths(path))
    if present:
        raise ContractError(
            '入力にサイドカーファイルが残っています: '
            + ','.join(candidate.name for candidate in present)
        )


def clone_read_only(source: Path, destination: Path) -> None:
    source_uri = source.as_uri() + '?mode=ro&immutable=1'
    source_connection = sqlite3.connect(source_uri, uri=True)
    try:
        source_connection.execute('PRAGMA query_only = ON')
        destination_connection = sqlite3.connect(destination)
        try:
            source_connection.backup(destination_connection)
        finally:
            destination_connection.close()
    finally:
        source_connection.close()


def expect_integrity_error(
    connection: sqlite3.Connection,
    label: str,
    sql: str,
    parameters: tuple[object, ...],
) -> None:
    rejected = False
    connection.execute('SAVEPOINT expected_failure')
    try:
        connection.execute(sql, parameters)
    except sqlite3.IntegrityError:
        rejected = True
    finally:
        connection.execute('ROLLBACK TO expected_failure')
        connection.execute('RELEASE expected_failure')

    if not rejected:
        raise ContractError(f'{label}が拒否されませんでした')


def check_integrity(connection: sqlite3.Connection) -> None:
    result = connection.execute(
        'PRAGMA integrity_check(1)'
    ).fetchone()
    if result != ('ok',):
        raise ContractError('integrity_checkを通過しませんでした')

    violation = connection.execute(
        'PRAGMA foreign_key_check'
    ).fetchone()
    if violation is not None:
        raise ContractError('外部キー違反があります')


def run_behavior_test(scratch: Path) -> tuple[int, list[str]]:
    connection = sqlite3.connect(scratch, isolation_level=None)
    try:
        connection.execute('PRAGMA foreign_keys = ON')
        foreign_keys = connection.execute(
            'PRAGMA foreign_keys'
        ).fetchone()
        if foreign_keys != (1,):
            raise ContractError('外部キー制約を有効にできません')

        journal = connection.execute(
            'PRAGMA journal_mode = DELETE'
        ).fetchone()
        if journal is None or str(journal[0]).lower() != 'delete':
            raise ContractError('一時DBをDELETEモードへ変更できません')

        version = int(
            connection.execute('PRAGMA user_version').fetchone()[0]
        )
        if version != EXPECTED_VERSION:
            raise ContractError(
                f'user_version が{EXPECTED_VERSION}ではありません: '
                f'{version}'
            )

        columns = {
            row[1]
            for row in connection.execute(
                "PRAGMA table_xinfo('review_items')"
            )
        }
        missing = sorted(REQUIRED_COLUMNS - columns)
        if missing:
            raise ContractError(
                '必須列がありません: ' + ','.join(missing)
            )

        check_integrity(connection)
        before_count = int(
            connection.execute(
                'SELECT count(*) FROM review_items'
            ).fetchone()[0]
        )

        for _ in range(5):
            token = secrets.token_hex(16)
            batch_id = f'__smoke__{token}'
            record_id = 'record'
            exists = connection.execute(
                '''
                SELECT 1
                  FROM review_items
                 WHERE batch_id = ? AND record_id = ?
                ''',
                (batch_id, record_id),
            ).fetchone()
            if exists is None:
                break
        else:
            raise ContractError('試験用IDを確保できませんでした')

        connection.execute('BEGIN IMMEDIATE')
        try:
            connection.execute(
                '''
                INSERT INTO review_items (
                    batch_id,
                    record_id,
                    proposed_category,
                    status
                ) VALUES (?, ?, ?, 'pending')
                ''',
                (batch_id, record_id, 'smoke-category'),
            )

            row = connection.execute(
                '''
                SELECT status, priority, decided_category
                  FROM review_items
                 WHERE batch_id = ? AND record_id = ?
                ''',
                (batch_id, record_id),
            ).fetchone()
            if row != ('pending', 0, None):
                raise ContractError('登録時の既定値が契約と異なります')

            changed = connection.execute(
                '''
                UPDATE review_items
                   SET priority = 3
                 WHERE batch_id = ? AND record_id = ?
                ''',
                (batch_id, record_id),
            ).rowcount
            if changed != 1:
                raise ContractError('優先度を1件更新できませんでした')

            expect_integrity_error(
                connection,
                'priority範囲外の更新',
                '''
                UPDATE review_items
                   SET priority = 4
                 WHERE batch_id = ? AND record_id = ?
                ''',
                (batch_id, record_id),
            )

            expect_integrity_error(
                connection,
                '主キーが重複する登録',
                '''
                INSERT INTO review_items (
                    batch_id,
                    record_id,
                    proposed_category,
                    status
                ) VALUES (?, ?, ?, 'pending')
                ''',
                (batch_id, record_id, 'smoke-category'),
            )

            expect_integrity_error(
                connection,
                '確定カテゴリのないacceptedへの遷移',
                '''
                UPDATE review_items
                   SET status = 'accepted'
                 WHERE batch_id = ? AND record_id = ?
                ''',
                (batch_id, record_id),
            )

            changed = connection.execute(
                '''
                UPDATE review_items
                   SET status = 'accepted',
                       decided_category = proposed_category
                 WHERE batch_id = ? AND record_id = ?
                ''',
                (batch_id, record_id),
            ).rowcount
            if changed != 1:
                raise ContractError('正常な状態遷移を実行できませんでした')

            row = connection.execute(
                '''
                SELECT status, priority, decided_category
                  FROM review_items
                 WHERE batch_id = ? AND record_id = ?
                ''',
                (batch_id, record_id),
            ).fetchone()
            if row != ('accepted', 3, 'smoke-category'):
                raise ContractError('状態遷移後の値が契約と異なります')
        finally:
            connection.rollback()

        after_count = int(
            connection.execute(
                'SELECT count(*) FROM review_items'
            ).fetchone()[0]
        )
        residue = connection.execute(
            '''
            SELECT 1
              FROM review_items
             WHERE batch_id = ? AND record_id = ?
            ''',
            (batch_id, record_id),
        ).fetchone()
        if after_count != before_count or residue is not None:
            raise ContractError('ロールバック後に試験データが残りました')

        check_integrity(connection)
    finally:
        connection.close()

    return version, [
        'foreign_keys_enabled',
        'version_and_required_columns',
        'valid_insert_priority_and_accept',
        'priority_constraint',
        'primary_key_constraint',
        'state_constraint',
        'rollback_without_residue',
        'integrity_and_foreign_keys',
    ]


def test_database(path: Path) -> dict[str, object]:
    require_sqlite()
    if not path.is_file():
        raise ContractError(f'入力DBが見つかりません: {path}')

    source = path.resolve(strict=True)
    reject_sidecars(source)
    digest_before = sha256_file(source)

    with tempfile.TemporaryDirectory(prefix='sqlite-smoke-') as directory:
        scratch = Path(directory) / 'scratch.sqlite'
        clone_read_only(source, scratch)

        if sha256_file(source) != digest_before:
            raise ContractError('複製中に入力DBの内容が変わりました')

        version, behavior_checks = run_behavior_test(scratch)

        reject_sidecars(source)
        if sha256_file(source) != digest_before:
            raise ContractError('試験中に入力DBの内容が変わりました')

    return {
        'status': 'passed',
        'command': 'test',
        'database': source.name,
        'user_version': version,
        'checks': [
            'snapshot_created',
            *behavior_checks,
            'source_hash_unchanged',
        ],
    }


def build_parser() -> JsonArgumentParser:
    parser = JsonArgumentParser(
        description='移行済みSQLiteの破棄可能な書き込み試験'
    )
    commands = parser.add_subparsers(dest='command', required=True)

    sample = commands.add_parser(
        'make-sample',
        help='上書きせずに動作確認用DBを作る',
    )
    sample.add_argument('database', type=Path)

    test = commands.add_parser(
        'test',
        help='入力を一時領域へ複製して試験する',
    )
    test.add_argument('database', type=Path)
    return parser


def main() -> int:
    arguments = build_parser().parse_args()
    try:
        if arguments.command == 'make-sample':
            result = make_sample(arguments.database)
        else:
            result = test_database(arguments.database)
    except KeyboardInterrupt:
        json_line(
            {'status': 'error', 'kind': 'interrupted'},
            sys.stderr,
        )
        return 130
    except ContractError as exc:
        json_line(
            {
                'status': 'error',
                'kind': 'contract',
                'message': str(exc),
            },
            sys.stderr,
        )
        return 2
    except (OSError, sqlite3.Error) as exc:
        json_line(
            {
                'status': 'error',
                'kind': 'runtime',
                'message': str(exc),
            },
            sys.stderr,
        )
        return 3
    except Exception as exc:
        json_line(
            {
                'status': 'error',
                'kind': 'unexpected',
                'message': f'{type(exc).__name__}: {exc}',
            },
            sys.stderr,
        )
        return 3

    json_line(result, sys.stdout)
    return 0


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

サンプル作成では、最初にDB本体と三種類のサイドカー名を検査し、DB本体をO_EXCLで排他的に作成します。作成中はディスク上のロールバックジャーナルを必要としないMEMORYモードを使い、接続を閉じた後にもサイドカーがないことを確認します。

作成後のDB本体が、排他的作成時に記録したデバイス番号とinode番号と一致しなければ成功にしません。失敗時も同じ照合を通過したDB本体だけを削除するため、以前からあったサイドカーを後始末の対象に含めません。

制約違反の確認では、各SQLをセーブポイント内で実行します。予定どおり拒否された場合だけでなく、誤って成功した場合にも、そのSQLを取り消してからテストを失敗させます。最後に試験全体のトランザクションも取り消します。

動作確認用データベースで実行する

最初に、記事内の契約と一致するサンプルを新しいファイルとして作ります。candidate.sqlite本体またはcandidate.sqlite-wal、candidate.sqlite-shm、candidate.sqlite-journalのいずれかが既にある場合、make-sampleは削除も上書きもせず、終了コード2で停止します。

python smoke_review_db.py make-sample candidate.sqlite

成功時の出力は次の1行です。

{"status":"created","command":"make-sample","database":"candidate.sqlite","user_version":2,"rows":1}

続いて、候補データベースを一時領域へ複製して試験します。

python smoke_review_db.py test candidate.sqlite

すべての検査を通過すると、次の1行が返ります。

{"status":"passed","command":"test","database":"candidate.sqlite","user_version":2,"checks":["snapshot_created","foreign_keys_enabled","version_and_required_columns","valid_insert_priority_and_accept","priority_constraint","primary_key_constraint","state_constraint","rollback_without_residue","integrity_and_foreign_keys","source_hash_unchanged"]}
終了コード 意味
0 サンプル作成またはスモークテストが完了した
2 引数、スキーマ、制約、既存の出力、サイドカーファイルなどが契約を満たさない
3 ファイル操作、SQLiteの実行、または予期しない処理で失敗した
130 Ctrl+Cで中断した

既存サイドカーを拒否することを確認する

次の例では空のblocked.sqlite-walを先に作ります。make-sampleはDB本体を作成せず、既存のサイドカーも削除しません。

python -c "open('blocked.sqlite-wal', 'xb').close()"
python smoke_review_db.py make-sample blocked.sqlite

標準エラー出力へ次のJSONを返し、終了コード2で停止します。

{"status":"error","kind":"contract","message":"出力先に既存ファイルがあります: blocked.sqlite-wal"}

この空ファイルは拒否動作だけを確認するためのものです。実際の候補DBに付随するサイドカーを、この手順で作成または削除してはいけません。

バージョン不一致を確認する

別のサンプルを作り、検査対象のuser_versionだけを1へ変更します。

python smoke_review_db.py make-sample wrong.sqlite
python -c "import sqlite3; db=sqlite3.connect('wrong.sqlite'); db.execute('PRAGMA user_version=1'); db.commit(); db.close()"
python smoke_review_db.py test wrong.sqlite

testは書き込み試験へ進まず、標準エラー出力へ次のJSONを返して終了コード2で停止します。

{"status":"error","kind":"contract","message":"user_version が2ではありません: 1"}

実際の移行済みDBへ合わせる箇所

  • EXPECTED_VERSIONを切り替え対象のスキーマバージョンへ合わせる
  • REQUIRED_COLUMNSと登録・更新・参照SQLを実際の列名へ置き換える
  • 成功させる操作と拒否すべき操作を業務上の状態遷移に合わせる
  • 外部キーの親データが必要なら、試験用の親行も同じトランザクション内で登録し、最後に取り消す
  • アプリケーションと同じSQLite実装を確認する場合は、そのアプリケーションと同じPython環境で実行する

実在する顧客IDや担当者IDを推測して試験値に使ってはいけません。専用の試験値を同じトランザクション内で作れないスキーマでは、匿名化した検証用データベースを別途用意します。

合格結果が保証しないこと

このスクリプトが確認するのは、現在のPythonに組み込まれたSQLiteで、指定したSQLと制約が一時コピー上で動くことです。アプリケーションの画面、認証、外部API、添付データベース、同時実行時の待ち時間までは確認しません。

入力ファイルのSHA-256比較は、試験中の内容変更を検出するための確認です。書き込みを排除するロックでも、改ざんを証明する仕組みでもありません。

immutable=1を指定すると、SQLiteは入力が変化しないものとして扱い、ロックや変更検出を省略します。入力が実際に変化すると、正しくない結果を返す可能性があります。実行前に書き込み元を停止し、サイドカーファイルがないことを確認する運用が前提です。

また、移行後のデータベースへ新しい業務データを書き込んだ後で古いファイルへ戻すと、その間の更新が失われます。今回のスモークテストは切り替え前の受け入れ判定であり、切り替え後の逆移行手順ではありません。

仕様を確認する一次資料

次の一歩

実DBへ適用する前に、専用の検証コピーでSQLと試験値を調整します。内部の正常操作と拒否検査がすべて期待どおりなら、testコマンド全体が終了コード0になります。

本番切り替え手順には、書き込み停止、最終バックアップ、移行、スモークテスト、切り替え可否の判断を別々の工程として記録します。

第167回 SQLiteのスキーマ変更を本番前に試す:Pythonで作るコピー移行と失敗時の破棄

第166回では、SQLiteバックアップを隔離環境へ復元し、ハッシュ、整合性、外部キー、スキーマバージョンを検査しました。復元できることを確認したら、スキーマ変更も本番データベースへ直接適用せず、コピー上で試します。

AIレビュー項目へ優先度を追加するときは、入力のデータベースを直接変更しないことが大切です。入力は読み取り専用で開き、新しいSQLiteファイルへコピーします。そのコピーに対して、列追加、インデックス作成、バージョン更新を一つのトランザクションで実行します。

サンプルのテーブルは動作確認用です。第163回以降のデータベースへそのまま適用するパッチではありません。実データへ使う場合は、列定義とDDLを実際のスキーマに合わせ、第166回で作ったような隔離コピーで検証してください。

移行の契約を固定する

項目 仕様
入力 PRAGMA user_versionが1で、所定のreview_items列を持つSQLite
入力の扱い 更新を止めた隔離コピーをmode=roで開き、SQLによる書き込みを行わない
出力 実行前には存在しない新しいSQLiteファイル。既存ファイルは上書きしない
変更内容 0から3までのpriority列と、未処理項目を優先順に読むための部分インデックスを追加する
既存行 業務上の優先度を推測せず、すべて0から開始する
移行単位 列追加、インデックス作成、user_version=2を一つのトランザクションで確定する
検査 列定義、インデックス定義、行数、初期優先度、整合性、外部キーを確認する
失敗時 トランザクションを取り消し、作成した出力ファイルを削除する
対象外 アプリケーションの切り替え、優先度の業務判定、同時マイグレーション、v2からv1への逆変換

動作環境はPython 3.10以上、SQLite 3.37.0以上とします。SQLite 3.37.0では、ADD COLUMNで追加するCHECK制約を既存行に対して検査する動作が導入されました。詳細はSQLiteのALTER TABLE仕様で確認できます。

Pythonが使用するSQLiteの版は、次のコマンドで確認します。

python3 -c "import sqlite3; print(sqlite3.sqlite_version)"

コピーとトランザクションの役割

新しいファイルへコピーするのは、移行内容に誤りがあっても入力を残すためです。トランザクションは、出力側で列だけが追加され、インデックスやバージョン更新が欠けた状態を確定させないために使います。

user_versionはアプリケーションが管理する整数であり、SQLiteが自動更新する値ではありません。値が1であることだけを信用せず、PRAGMA table_xinfoで列名、型、NULL制約、既定値、主キー、隠し列の区分を照合します。user_versionの仕様はSQLiteのPRAGMA文書で確認できます。

1.動作確認用のv1データベースを作る

次のコードをcreate_review_v1.pyとして保存します。2件のデータは移行結果を確かめるためのサンプルです。

from __future__ import annotations

import json
import os
import sqlite3
from contextlib import closing
from pathlib import Path

path = Path("review_v1.sqlite")
try:
    fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
except FileExistsError:
    raise SystemExit("review_v1.sqlite already exists")
os.close(fd)

completed = False
try:
    with closing(sqlite3.connect(path)) as conn:
        conn.execute("PRAGMA foreign_keys = ON")
        conn.execute(
            """
            CREATE TABLE review_items (
                item_id INTEGER PRIMARY KEY,
                record_id TEXT NOT NULL UNIQUE,
                status TEXT NOT NULL
                    CHECK (status IN ('pending', 'accepted', 'corrected', 'rejected')),
                created_at TEXT NOT NULL
            )
            """
        )
        conn.executemany(
            """
            INSERT INTO review_items(record_id, status, created_at)
            VALUES (?, ?, ?)
            """,
            [
                ("R-001", "pending", "2026-01-10T09:00:00Z"),
                ("R-002", "accepted", "2026-01-10T09:05:00Z"),
            ],
        )
        conn.execute("PRAGMA user_version = 1")
        conn.commit()
    completed = True
finally:
    if not completed:
        path.unlink(missing_ok=True)

print(
    json.dumps(
        {"status": "created", "database": str(path), "rows": 2},
        ensure_ascii=False,
        separators=(",", ":"),
    )
)

実行すると、新しいreview_v1.sqliteが作られます。同名ファイルがある場合は上書きしません。

python3 create_review_v1.py
{"status":"created","database":"review_v1.sqlite","rows":2}

2.コピー上でv1からv2へ移行する

次のコードをmigrate_review_db.pyとして保存します。標準ライブラリ以外は使用しません。

from __future__ import annotations

import argparse
import json
import os
import sqlite3
import sys
from contextlib import closing
from pathlib import Path
from typing import Any, TextIO

FROM_VERSION = 1
TO_VERSION = 2
MIN_SQLITE = (3, 37, 0)
INDEX_NAME = "ix_review_items_pending_priority"
INDEX_SQL = """
CREATE INDEX ix_review_items_pending_priority
ON review_items(priority DESC, created_at, item_id)
WHERE status = 'pending'
"""

V1_COLUMNS = [
    ("item_id", "INTEGER", 0, None, 1, 0),
    ("record_id", "TEXT", 1, None, 0, 0),
    ("status", "TEXT", 1, None, 0, 0),
    ("created_at", "TEXT", 1, None, 0, 0),
]
V2_COLUMNS = V1_COLUMNS + [
    ("priority", "INTEGER", 1, "0", 0, 0),
]


class ContractError(Exception):
    pass


def emit(stream: TextIO, payload: dict[str, Any]) -> None:
    print(
        json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
        file=stream,
    )


class JsonParser(argparse.ArgumentParser):
    def error(self, message: str) -> None:
        emit(sys.stderr, {"status": "rejected", "message": message})
        raise SystemExit(2)


def readonly_uri(path: Path) -> str:
    return path.as_uri() + "?mode=ro"


def column_signature(conn: sqlite3.Connection) -> list[tuple[Any, ...]]:
    rows = conn.execute("PRAGMA table_xinfo('review_items')").fetchall()
    return [
        (
            str(row[1]),
            str(row[2]).upper(),
            int(row[3]),
            row[4],
            int(row[5]),
            int(row[6]),
        )
        for row in rows
    ]


def normalize_sql(sql: str) -> str:
    return " ".join(sql.split())


def validate_contract(conn: sqlite3.Connection, version: int) -> None:
    actual_version = int(conn.execute("PRAGMA user_version").fetchone()[0])
    if actual_version != version:
        raise ContractError(
            f"user_versionが{version}ではありません: {actual_version}"
        )

    expected = V1_COLUMNS if version == FROM_VERSION else V2_COLUMNS
    if column_signature(conn) != expected:
        raise ContractError(f"review_itemsの列定義がv{version}契約と一致しません")

    index_row = conn.execute(
        """
        SELECT tbl_name, sql
        FROM sqlite_schema
        WHERE type = 'index' AND name = ?
        """,
        (INDEX_NAME,),
    ).fetchone()

    if version == FROM_VERSION and index_row is not None:
        raise ContractError(f"v1に予約済みインデックス名があります: {INDEX_NAME}")

    if version == TO_VERSION:
        if (
            index_row is None
            or index_row[0] != "review_items"
            or index_row[1] is None
            or normalize_sql(str(index_row[1])) != normalize_sql(INDEX_SQL)
        ):
            raise ContractError("v2の優先度インデックス定義が一致しません")


def check_database(conn: sqlite3.Connection) -> None:
    integrity = conn.execute("PRAGMA integrity_check(1)").fetchone()
    if integrity is None or str(integrity[0]).lower() != "ok":
        raise ContractError("SQLiteの整合性検査に合格しませんでした")

    if conn.execute("PRAGMA foreign_key_check").fetchone() is not None:
        raise ContractError("外部キー違反があります")


def row_count(conn: sqlite3.Connection) -> int:
    return int(conn.execute("SELECT COUNT(*) FROM review_items").fetchone()[0])


def copy_database(source: Path, target: Path) -> None:
    with closing(
        sqlite3.connect(readonly_uri(source), uri=True, timeout=5.0)
    ) as src:
        src.execute("PRAGMA query_only = ON")
        with closing(sqlite3.connect(target, timeout=5.0)) as dst:
            src.backup(dst)


def migrate_copy(target: Path) -> int:
    with closing(
        sqlite3.connect(target, timeout=5.0, isolation_level=None)
    ) as conn:
        conn.execute("PRAGMA foreign_keys = ON")
        journal = conn.execute("PRAGMA journal_mode = DELETE").fetchone()
        if journal is None or str(journal[0]).lower() != "delete":
            raise ContractError("出力をDELETEジャーナルモードへ変更できません")

        validate_contract(conn, FROM_VERSION)
        check_database(conn)
        before = row_count(conn)

        conn.execute("BEGIN IMMEDIATE")
        try:
            conn.execute(
                """
                ALTER TABLE review_items
                ADD COLUMN priority INTEGER NOT NULL DEFAULT 0
                    CHECK (priority BETWEEN 0 AND 3)
                """
            )
            conn.execute(INDEX_SQL)
            conn.execute("PRAGMA user_version = 2")

            if row_count(conn) != before:
                raise ContractError("移行中にreview_itemsの行数が変わりました")

            invalid_priority = int(
                conn.execute(
                    """
                    SELECT COUNT(*)
                    FROM review_items
                    WHERE priority IS NULL OR priority <> 0
                    """
                ).fetchone()[0]
            )
            if invalid_priority != 0:
                raise ContractError("既存行のpriorityが0で初期化されていません")

            validate_contract(conn, TO_VERSION)
            check_database(conn)
            conn.commit()
        except (Exception, KeyboardInterrupt):
            conn.rollback()
            raise

    return before


def verify_output(target: Path, expected_rows: int) -> None:
    with closing(sqlite3.connect(readonly_uri(target), uri=True)) as conn:
        conn.execute("PRAGMA foreign_keys = ON")
        conn.execute("PRAGMA query_only = ON")
        validate_contract(conn, TO_VERSION)
        check_database(conn)

        if row_count(conn) != expected_rows:
            raise ContractError("再オープン後の行数が一致しません")

        invalid_priority = int(
            conn.execute(
                """
                SELECT COUNT(*)
                FROM review_items
                WHERE priority IS NULL OR priority <> 0
                """
            ).fetchone()[0]
        )
        if invalid_priority != 0:
            raise ContractError("再オープン後のpriority検査に失敗しました")


def resolve_paths(source_text: str, target_text: str) -> tuple[Path, Path]:
    try:
        source = Path(source_text).expanduser().resolve(strict=True)
    except (OSError, RuntimeError) as exc:
        raise ContractError(f"入力を解決できません: {source_text}") from exc

    if not source.is_file():
        raise ContractError(f"入力は通常ファイルではありません: {source_text}")

    target_input = Path(target_text).expanduser()
    if os.path.lexists(target_input):
        raise ContractError(f"出力先は既に存在します: {target_text}")

    try:
        target = target_input.resolve(strict=False)
    except (OSError, RuntimeError) as exc:
        raise ContractError(f"出力先を解決できません: {target_text}") from exc

    if source == target:
        raise ContractError("入力と出力に同じパスは指定できません")
    if not target.parent.is_dir():
        raise ContractError(f"出力先ディレクトリがありません: {target.parent}")

    return source, target


def migrate(source_text: str, target_text: str) -> dict[str, Any]:
    if sqlite3.sqlite_version_info < MIN_SQLITE:
        raise ContractError(
            f"SQLite 3.37.0以上が必要です: {sqlite3.sqlite_version}"
        )

    source, target = resolve_paths(source_text, target_text)
    created = False
    try:
        try:
            fd = os.open(target, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
        except FileExistsError as exc:
            raise ContractError(f"出力先は既に存在します: {target_text}") from exc

        created = True
        os.close(fd)
        copy_database(source, target)
        rows = migrate_copy(target)
        verify_output(target, rows)
    except (Exception, KeyboardInterrupt):
        if created:
            target.unlink(missing_ok=True)
        raise

    return {
        "status": "migrated",
        "from_version": FROM_VERSION,
        "to_version": TO_VERSION,
        "rows": rows,
        "output": target_text,
    }


def main() -> int:
    parser = JsonParser(description="SQLiteのv1コピーをv2へ移行します")
    parser.add_argument("source", help="読み取り元のv1 SQLite")
    parser.add_argument("output", help="存在しない出力先")
    args = parser.parse_args()

    try:
        report = migrate(args.source, args.output)
    except ContractError as exc:
        emit(sys.stderr, {"status": "rejected", "message": str(exc)})
        return 2
    except KeyboardInterrupt:
        emit(sys.stderr, {"status": "interrupted"})
        return 130
    except Exception as exc:
        emit(
            sys.stderr,
            {
                "status": "error",
                "error_type": type(exc).__name__,
                "message": str(exc),
            },
        )
        return 3

    emit(sys.stdout, report)
    return 0


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

3.マイグレーションを実行する

入力とは異なる、まだ存在しない出力先を指定します。

python3 migrate_review_db.py review_v1.sqlite review_v2.sqlite

成功時は標準出力へ次の1行を返し、終了コードは0です。

{"status":"migrated","from_version":1,"to_version":2,"rows":2,"output":"review_v2.sqlite"}

出力を読み取り専用で開き、バージョンと既存行の優先度を確認します。

python3 -c "import sqlite3; c=sqlite3.connect('file:review_v2.sqlite?mode=ro', uri=True); print(c.execute('PRAGMA user_version').fetchone()[0]); print(c.execute('SELECT record_id, status, priority FROM review_items ORDER BY item_id').fetchall()); c.close()"
2
[('R-001', 'pending', 0), ('R-002', 'accepted', 0)]

元のreview_v1.sqliteはバージョン1のままです。既存行へ高い優先度を自動設定しないため、業務担当者が確認した後の更新処理は別に用意します。

4.誤ったバージョンを拒否する

作成したv2を、もう一度v1入力として渡します。スクリプトはバージョン不一致を検出し、終了コード2を返します。作成した出力ファイルも削除します。

python3 migrate_review_db.py review_v2.sqlite rejected.sqlite
echo $?
test ! -e rejected.sqlite && echo absent
{"status":"rejected","message":"user_versionが1ではありません: 2"}
2
absent

すでに存在する出力先を指定した場合も終了コード2で停止します。事前確認に加えて、os.open()による排他的作成を使うため、確認直後に同名ファイルが作られても上書きしません。

終了コード

終了コード 意味 後続処理
0 コピー、移行、再オープン後の検査が完了 アプリケーション側の読み書き試験へ進む
2 バージョン、列定義、出力先などが契約外 入力と移行手順を確認する
3 SQLite、ファイル操作、または予期しない実行エラー エラー種別と実行環境を確認する
130 Ctrl+Cによる中断 出力が残っていないことを確認してから再実行する

--helpは通常のヘルプ文を表示します。それ以外にスクリプトが処理した結果は、標準出力または標準エラー出力へ1行のJSONとして返します。

この実装で保証しないこと

  • 移行の成功は、アプリケーションがv2へ対応している証明ではない。実際に使用するSELECT、INSERT、UPDATEを隔離環境で試す。
  • 部分インデックスを作成しても、既存の取得処理が自動的に優先順へ変わるわけではない。取得SQLにもWHERE status = 'pending'とORDER BY priority DESC, created_at, item_idが必要になる。
  • created_atはTEXTとして保持するだけで、日時書式や時系列上の正しさまでは検査しない。
  • 入力は更新されていない隔離コピーとし、出力先も別プロセスが操作しない、権限を制御したローカルディレクトリを使う。
  • 強制終了、OS停止、電源断ではPythonの後始末が動かず、不完全な出力が残る場合がある。終了コード0と成功JSONを確認できないファイルは採用しない。
  • v2へ新しい業務データを書き込んだ後にv1ファイルへ戻しても、追加データは引き継がれない。切り戻し条件とデータ変換は本番切り替え前に決める。

実データへ適用する前の確認

  1. 実際のreview_itemsに合わせて、v1とv2の列定義を固定する。
  2. 追加列の既定値を推測で決めず、既存行へ安全に適用できる値を選ぶ。
  3. 検証済みバックアップから作った、更新されない隔離データベースを入力にする。
  4. 終了コード0、成功JSON、行数、整合性検査の結果を確認する。
  5. v2対応のアプリケーションで、読み取りと書き込みのスモークテストを行う。
  6. 入力となったv1コピーは、受け入れ確認が終わるまで変更せずに保管する。

コピー作成、契約検査、トランザクション、再オープン後の検査を分けると、DDLを実行できたことと、業務で採用できることを別々に判断できます。