物理サーバーで SQLite のオンラインバックアップを検証する

バックアップジョブは成功と表示されているのに、復元すると直近のレコードが欠けている。SQLite を WAL モードで使うファイルサービスでは、こうした不具合を見誤りがちです。ファイルのコピーが成功しても、それはバイト列の転送が完了したことを示すだけで、データベースの整合性までは保証しません。ファイルのインデックス、ジョブの状態、バックアップ一覧などを継続的に記録する物理サーバーでは、「コピーを作ること」と「そのコピーが使えることを確かめること」を別の工程として扱う必要があります。

1. WAL とバックアップの役割を整理する

WAL モードでは、コミット済みの変更がまず -wal ファイルに保存され、その後チェックポイントによってデータベース本体に書き戻されることがあります。サービスの稼働中に cp service.db backup.db を実行すると、こうした変更を取りこぼす可能性があります。本体、WAL、共有メモリファイルを順番にコピーしても、同一時点のスナップショットを取得したことにはなりません。

オンラインバックアップ API は、変化し続けるファイルをそのままコピーするのではなく、データベース接続を通じて整合性のある内容を読み取ります。書き込みを停止できない用途に適していますが、読み取りとコピー先への書き込みによる負荷は発生します。すべての接続を閉じられるなら、オフラインバックアップのほうが単純です。ただし、書き込みを止める際は、フロントエンドからのリクエストだけでなく、バックグラウンドジョブも停止対象に含める必要があります。

WAL は遠隔地バックアップではなく、ミラーリングもバックアップではありません。信頼できるコピーには、元のデータベースとは異なる障害範囲での保管と、独立した読み取り検証が必要です。

OnuCloud の物理サーバーには 1TB NVMe が 2 台標準搭載されており、それぞれ直接接続され、ハードウェア RAID はありません。もう一方のドライブにコピーを書き出せば、読み書きの負荷を一部分離できます。ただし、どちらのドライブも同じサーバー内にあるため、遠隔地での保管の代わりにはなりません。

2. 継続的に書き込む検証データを再現可能な形で用意する

以下の検証は、Debian 12 など、Python 3 と標準ライブラリの sqlite3 モジュールが使える環境を対象としています。まずモジュールを確認し、専用の一時ディレクトリを作成します。操作の対象はすべて検証用ファイルに限定され、本番データベースには触れず、システムサービスの変更も不要です。

python3 -c 'import sqlite3; print(sqlite3.sqlite_version)'
umask 077
export LAB_DIR="$(mktemp -d)"
printf '%s\n' "$LAB_DIR"

検証用データベースには 2 つのテーブルを使います。events に明細を、totals に累計値を保存します。各トランザクションで明細の追加と累計値の加算を同時に行うため、コピーは「明細の金額の合計と累計値が等しい」という条件を満たす必要があります。この条件は、ファイルの存在確認よりも意味があり、構造の整合性と業務上の正しさが別々の検証項目であることも示せます。

以下のスクリプトには、書き込みスレッドとバックアップ処理の両方が含まれています。使用するディレクトリは前の手順から引き継ぎます。古い検証データと混在させないよう、再検証するたびにディレクトリを作り直してください。本番データベースは通常これより複雑なので、テーブル間の関係やアプリケーションのバージョン要件を別途定義する必要があります。このカウントモデルをそのまま当てはめることはできません。

3. オンラインバックアップ API でコピーを作成し、検証する

スクリプトは、最初のコミットが完了した後も書き込みスレッドを動かし続け、その間にオンラインバックアップを実行します。出力先にはまず一時ファイルを使い、整合性チェック、外部キーチェック、業務上の不変条件の確認がすべて通った場合にだけ、検証済みの正式なコピー名に変更します。どの段階で異常が発生してもスクリプトは失敗し、新しいコピーは公開されません。

python3 - <<'PY'
import os
import sqlite3
import threading
import time
from pathlib import Path

base = Path(os.environ["LAB_DIR"])
source = base / "source.db"
pending = base / "backup.pending.db"
published = base / "backup.db"

db = sqlite3.connect(source)
mode = db.execute("PRAGMA journal_mode=WAL").fetchone()[0]
if mode.lower() != "wal":
    raise RuntimeError("WAL mode unavailable")
db.executescript("""
CREATE TABLE events (
    n INTEGER PRIMARY KEY,
    amount INTEGER NOT NULL
);
CREATE TABLE totals (
    id INTEGER PRIMARY KEY CHECK (id = 1),
    balance INTEGER NOT NULL
);
INSERT INTO totals VALUES (1, 0);
""")
db.close()

stop = threading.Event()
ready = threading.Event()
errors = []

def writer():
    conn = None
    try:
        conn = sqlite3.connect(source, timeout=5)
        while not stop.is_set():
            with conn:
                conn.execute("INSERT INTO events(amount) VALUES (1)")
                conn.execute(
                    "UPDATE totals SET balance = balance + 1 WHERE id = 1"
                )
            ready.set()
            time.sleep(0.01)
    except Exception as exc:
        errors.append(exc)
        ready.set()
    finally:
        if conn is not None:
            conn.close()

worker = threading.Thread(target=writer)
worker.start()
try:
    if not ready.wait(5):
        raise RuntimeError("Writer did not become ready")
    if errors:
        raise RuntimeError("Writer failed") from errors[0]
    src = sqlite3.connect(source, timeout=5)
    dst = sqlite3.connect(pending, timeout=5)
    try:
        src.backup(dst, pages=128, sleep=0.01)
    finally:
        dst.close()
        src.close()
finally:
    stop.set()
    worker.join()

if errors:
    raise RuntimeError("Writer failed") from errors[0]

check = sqlite3.connect(pending)
try:
    result = check.execute("PRAGMA integrity_check").fetchall()
    if result != [("ok",)]:
        raise RuntimeError("Integrity check failed")
    if check.execute("PRAGMA foreign_key_check").fetchall():
        raise RuntimeError("Foreign key check failed")
    count, amount = check.execute(
        "SELECT COUNT(*), COALESCE(SUM(amount), 0) FROM events"
    ).fetchone()
    balance = check.execute(
        "SELECT balance FROM totals WHERE id = 1"
    ).fetchone()[0]
    if not (count > 0 and count == amount == balance):
        raise RuntimeError("Business invariant failed")
finally:
    check.close()

os.replace(pending, published)
print(f"Verified rows: {count}")
print(published)
PY

pages=128 は、1 回の処理で扱うデータベースページ数であり、バイト数ではありません。sleep は主にビジー状態での再試行時に使われる値で、スループットを一定に制限するものではありません。本番運用では所要時間と失敗理由を記録し、ジョブ全体の制限時間を設けてください。高頻度の書き込みが続くとバックアップが長引く可能性があります。タイムアウトした場合は失敗と判定し、未完成のファイルを成功扱いにしてはいけません。

ここでのファイル名変更は、検証前の候補と検証済みのコピーを区別するためのものであり、電源断に対する永続性を保証するものではありません。本番のバックアップ処理では、ファイルの同期、ディレクトリの同期、保存先ストレージの動作仕様を踏まえ、どの時点でデータの永続化を保証するかも設計する必要があります。

4. 復元を隔離し、元のデータベースでコピーの正しさを証明しない

バックアップ完了後も元のデータベースでは書き込みが進んでいる可能性があるため、後から取得した元データベースの行数とコピーの行数が一致することを条件にはできません。まずコピー自体の整合性を確認し、バックアップの実行時間帯に基づいてデータの鮮度を判断してください。オンラインバックアップは継続的なレプリケーションではなく、任意の時点に復元する機能を自動的に提供するものでもありません。

正式なコピーから復元検証用の作業コピーを作る

検証用のバックアップはすでに接続が閉じられているため、別のディレクトリへオフラインでコピーできます。以下では復元検証用の作業コピーを読み取り専用で開き、テーブル構造と主要な関係を確認します。復元テストの接続先を source.db にしてはいけません。それでは元のデータベースを検証することになります。

mkdir "$LAB_DIR/restore"
cp "$LAB_DIR/backup.db" "$LAB_DIR/restore/service.db"
python3 - <<'PY'
import os
import sqlite3
from pathlib import Path

path = Path(os.environ["LAB_DIR"]) / "restore" / "service.db"
conn = sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)
try:
    print(conn.execute("PRAGMA integrity_check").fetchall())
    print(conn.execute("PRAGMA foreign_key_check").fetchall())
    print(conn.execute("""
        SELECT
            (SELECT COALESCE(SUM(amount), 0) FROM events),
            (SELECT balance FROM totals WHERE id = 1)
    """).fetchone())
finally:
    conn.close()
PY

合格条件は、整合性チェックの結果が ok であること、外部キーチェックで行が返されないこと、2 つの累計値が等しいことです。この検証用データベースには外部キーがないため、外部キーチェックの結果が空であることは、チェック処理を実行できたことしか示しません。実際の業務では、本来の制約を残した状態で検証してください。

本番の受け入れ検証では、対応するバージョンのアプリケーションで作業コピーを開き、ファイルのインデックスやジョブ一覧を読み取る必要もあります。その際は、メール送信、コールバック、自動実行ジョブを無効にしてください。データベースを開けるからといって、外部ファイルまでそろっているとは限りません。

5. 単発の検証をバックアップの受け入れ条件にする

本番導入前には、少なくとも 4 項目の証跡を残してください。元のデータベースのパスとアプリケーションのバージョン、バックアップの開始・終了時刻、チェック結果、隔離環境での読み取り結果です。時刻はデータの鮮度を判断するためのものであり、トランザクションがどこまで含まれるかを厳密に示すものではありません。明確な復元時点が求められる業務では、コミットの連番管理と復元目標を別途設計する必要があります。

バックアップディレクトリのアクセス権を制限し、データベース内のトークン、ユーザー情報、ジョブのパラメーターがコピー経由で漏えいしないようにしてください。遠隔地へ転送した後は、ファイルのチェックサムを照合し、復元テストを繰り返します。チェックサムの一致が示すのは転送の完全性だけであり、データベースの受け入れ検証の代わりにはなりません。

失敗時には、前回の検証済みコピーを保持し、検証前の候補ファイルは別に扱ってください。成功済みの記録を空のファイルで上書きしてはいけません。データベースが添付ファイル、インデックスのシャード、別のデータベースを参照している場合、単一データベースのオンラインバックアップでは、複数リソースにまたがる原子性は保証されません。業務側で書き込みを止めるか、バージョン一覧によって整合性を調整する必要があります。

最終的な受け入れ条件は「コピーコマンドが成功したこと」ではなく、「コピーが独立して存在すること、構造のチェックに合格すること、業務上の関係が成立すること、アプリケーションが隔離環境で読み取れること」です。この 4 項目がすべて満たされて初めて、復元可能であることを検証できるバックアップになります。

よくある質問