آزمون بازیابی پشتیبان SQLite؛ بازخوانی مستقل و مرز شواهد محلی

وجود یک فایل با نام پشتیبان، به‌تنهایی نشان نمی‌دهد بتوانیم دادهٔ مورد انتظار را از آن برگردانیم. حتی بازشدن موفق پایگاه داده هم لزوماً کافی نیست؛ ممکن است فایل سالم باشد ولی نسخه یا محتوای اشتباهی داشته باشد. این درس، پشتیبان واقعیِ یک پایگاه دادهٔ کوچکِ ساختگی را می‌سازد، اتصال مقصد را می‌بندد، آن را دوباره باز می‌کند و بازیابی را در محیطی جدا می‌آزماید. هیچ پایگاه دادهٔ کاربر، دستگاه حسگر یا خرابی واقعی هدف آزمایش نیست.

۱. معیار پذیرش را با فهرست منطقیِ معلوم شروع کنیم

دادهٔ مرجع دو دستگاه با نام‌های coil-A و rail-B و سه نمونه دارد. نمونهٔ ۱۰۱ برای دستگاه یک، زمان ۱۰۰۰ و مقدار ۱۲۰۰ میلی‌ولت دارد؛ نمونهٔ ۱۰۲ در زمان ۱۰۱۰ مقدار NULL دارد؛ نمونهٔ ۲۰۱ برای دستگاه دوم، زمان ۱۰۰۰ و مقدار منفی پنجاه دارد. NULL را صفر نمی‌گیریم و علامت منفی را حذف نمی‌کنیم. همهٔ این عددها ساختگی و زمان‌ها صحیحِ میلی‌ثانیه‌اند.

علاوه بر ردیف‌ها، واحد mV در جدول فراداده و user_version برابر هفت ثبت می‌شود. یک اندیسِ مشخص روی شناسهٔ دستگاه، زمان و شناسهٔ نمونه داریم. مرجعِ پذیرش، ساختار پایتونیِ ازپیش‌نوشته‌شده با همین ردیف‌هاست؛ آن را از خروجیِ دوباره‌خوانده‌شدهٔ SQL تولید نمی‌کنیم. نامِ نسخهٔ منطقیِ هفت نیز قرارداد همین fixture است، نه نسخهٔ SQLite یا تاریخِ فایل.

پرسش‌های پذیرش جدا هستند: آیا فایل دوباره باز می‌شود؟ آیا نام جدول‌ها و اندیس و ترتیب ستون‌های اندیس حاضر است؟ آیا ردیف، NULL، واحد و نسخه با مرجع می‌خوانند؟ آیا بررسی ساختاری خطایی یافته است؟ موفقیت در یکی، جواب بقیه را به‌طور خودکار نمی‌دهد. اندازهٔ غیرصفرِ فایل فقط یک مشاهدهٔ مقدماتی است و جای بازخوانیِ داده را نمی‌گیرد.

۲. کپی را از اتصال مبدأ به مقصدِ تازه انجام دهیم

Online Backup API در SQLite محتوای پایگاه داده را از مسیر خودِ موتور به مقصد دیگر کپی می‌کند. مقصد می‌تواند محتوای قبلی‌اش را از دست بدهد؛ برای همین تمام مقصدهای این تمرین تازه و خصوصی‌اند. این برنامه، دستورِ کپیِ خامِ فایلِ یک پایگاه دادهٔ زنده نیست و مسیرِ پایگاه دادهٔ موجودِ خواننده را هم دریافت نمی‌کند.

در رابط Connection.backup پایتون، جهت فراخوانی از مبدأ به اتصال هدف است. ابتدا دادهٔ مبدأ با COMMIT ثبت و نبود تراکنش نوشتنِ باز کنترل می‌شود؛ سپس backup فراخوانی می‌شود. نویسندهٔ هم‌زمان، رشتهٔ پس‌زمینه و وقفهٔ مصنوعی نداریم. قابلیت آنلاینِ API در منبع توضیح داده شده، ولی رقابتِ آنلاین در این آزمایش اجرا نشده است.

حالت journal مبدأ DELETE است و WAL فعال نیست. autocommit پایتون درست است و مرزهای BEGIN و COMMIT با دستور SQL صریح‌اند. پوشهٔ یکتا کنار اسکریپت ساخته و مسیر مطلقش کنترل می‌شود؛ سه نامِ مبدأ، پشتیبانِ پیشین و پشتیبانِ تازه جدا هستند. این جداسازی برای جلوگیری از دست‌کاریِ دادهٔ موجود در همین fixture است؛ سامانهٔ کنترل دسترسی یا نگهداری نسخه‌های تولید نیست.

۳. اتصال مقصد را ببندیم و مستقل بازخوانی کنیم

پس از برگشتِ backup، اتصالِ نوشتنیِ مقصد بسته می‌شود. اتصالِ تازه با mode=ro همان فایل را باز می‌کند و هر جدول با ORDER BY معلوم خوانده می‌شود. این کار، استفادهٔ دوباره از شیء اتصالِ درحال کپی نیست. تلاش آزمایشی برای DELETE در این خواننده با SQLITE_READONLY برابر هشت رد می‌شود؛ این نتیجه مربوط به بازکردنِ مشخصِ همان فایل است و رمزنگاری یا حفاظت از همهٔ مسیرهای دسترسی را ثابت نمی‌کند.

"""Actual private local SQLite backups and independent logical restore checks."""
import json
import platform
import sqlite3
import tempfile
from contextlib import ExitStack, closing
from pathlib import Path

DEVICES = [(1, "coil-A"), (2, "rail-B")]
SAMPLES = [(101, 1, 1000, 1200), (102, 1, 1010, None), (201, 2, 1000, -50)]
META = [("unit", "mV")]


def readback(connection):
    return {
        "devices": connection.execute("SELECT id,name FROM device ORDER BY id").fetchall(),
        "samples": connection.execute("SELECT id,device_id,ts_ms,voltage_mv FROM sample ORDER BY id").fetchall(),
        "metadata": connection.execute("SELECT key,value FROM meta ORDER BY key").fetchall(),
        "user_version": connection.execute("PRAGMA user_version").fetchone()[0],
    }


def expected(changed=False):
    # Hand-authored logical manifest, independently of backup/readback SQL output.
    samples = [(101, 1, 1000, 1250), (202, 2, 1020, 300)] if changed else list(SAMPLES)
    return {"devices": list(DEVICES), "samples": samples,
            "metadata": list(META), "user_version": 7}


def run_example():
    checks = {}
    def check(name, condition):
        checks[name] = bool(condition)
        if not checks[name]:
            raise AssertionError(name)

    parent = Path(__file__).resolve().parent
    folder = Path(tempfile.mkdtemp(prefix="sqlite-backup-fixture-", dir=parent)).resolve()
    if folder.parent != parent or not folder.name.startswith("sqlite-backup-fixture-"):
        raise RuntimeError("temporary workspace boundary")
    paths = [folder / name for name in ("source.sqlite", "backup-before.sqlite", "backup-after.sqlite")]
    if any(p.resolve().parent != folder for p in paths):
        raise RuntimeError("database path boundary")
    output = None
    try:
        with ExitStack() as stack:
            source = stack.enter_context(closing(sqlite3.connect(paths[0], autocommit=True)))
            source.execute("PRAGMA foreign_keys=ON")
            journal_mode = source.execute("PRAGMA journal_mode=DELETE").fetchone()[0]
            source.executescript("""
              BEGIN;
              CREATE TABLE device(id INTEGER PRIMARY KEY,name TEXT NOT NULL UNIQUE);
              CREATE TABLE sample(id INTEGER PRIMARY KEY,device_id INTEGER NOT NULL
                REFERENCES device(id),ts_ms INTEGER NOT NULL,voltage_mv INTEGER);
              CREATE TABLE meta(key TEXT PRIMARY KEY,value TEXT NOT NULL);
              CREATE INDEX sample_device_time ON sample(device_id,ts_ms,id);
              PRAGMA user_version=7;
              COMMIT;
            """)
            source.execute("BEGIN")
            source.executemany("INSERT INTO device VALUES(?,?)", DEVICES)
            source.executemany("INSERT INTO sample VALUES(?,?,?,?)", SAMPLES)
            source.executemany("INSERT INTO meta VALUES(?,?)", META)
            source.execute("COMMIT")
            check("committed_source_matches_hand_manifest", not source.in_transaction and readback(source) == expected())
            with closing(sqlite3.connect(paths[1], autocommit=True)) as target:
                source.backup(target)
            # The original target connection is closed before opening a new reader.
            reader = stack.enter_context(closing(sqlite3.connect(paths[1].as_uri() + "?mode=ro", uri=True, autocommit=True)))
            first_read = readback(reader)
            check("closed_then_reopened_backup_matches_manifest", paths[1].stat().st_size > 0 and first_read == expected())
            objects = reader.execute("SELECT type,name FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%' ORDER BY type,name").fetchall()
            index_columns = [r[2] for r in reader.execute("PRAGMA index_info(sample_device_time)")]
            check("tables_and_explicit_index_preserved", objects == [("index", "sample_device_time"), ("table", "device"), ("table", "meta"), ("table", "sample")] and index_columns == ["device_id", "ts_ms", "id"])
            check("metadata_signed_value_and_null_preserved", first_read["user_version"] == 7 and first_read["metadata"] == [("unit", "mV")] and first_read["samples"][1][3] is None and first_read["samples"][2][3] == -50)
            integrity = reader.execute("PRAGMA integrity_check").fetchall()
            check("backup_integrity_check_ok", integrity == [("ok",)])
            check("backup_foreign_key_check_separate_and_empty", reader.execute("PRAGMA foreign_key_check").fetchall() == [])
            try:
                reader.execute("DELETE FROM sample")
            except sqlite3.OperationalError as error:
                readonly_code = error.sqlite_errorcode
            else:
                raise AssertionError("read-only reader unexpectedly wrote")
            check("readonly_reopen_rejects_write", readonly_code == sqlite3.SQLITE_READONLY and readback(reader) == expected())
            source.executescript("""BEGIN;
              UPDATE sample SET voltage_mv=1250 WHERE id=101;
              DELETE FROM sample WHERE id IN(102,201);
              INSERT INTO sample VALUES(202,2,1020,300);
              COMMIT;""")
            check("later_source_changes_match_second_hand_manifest", readback(source) == expected(changed=True))
            check("old_backup_unchanged_after_later_source_commit", readback(reader) == expected())
            source.execute("BEGIN")
            source.execute("UPDATE sample SET voltage_mv=9999 WHERE id=101")
            source.execute("ROLLBACK")
            check("rollback_not_in_second_committed_manifest", not source.in_transaction and readback(source) == expected(changed=True))
            restored = stack.enter_context(closing(sqlite3.connect(":memory:", autocommit=True)))
            reader.backup(restored)
            check("restore_into_fresh_private_memory_matches_old_manifest", readback(restored) == expected())
            with closing(sqlite3.connect(paths[2], autocommit=True)) as target:
                source.backup(target)
            latest = stack.enter_context(closing(sqlite3.connect(paths[2].as_uri() + "?mode=ro", uri=True, autocommit=True)))
            check("second_backup_captures_new_state_without_replacing_first", readback(latest) == expected(changed=True) and readback(reader) == expected())
            # Structurally valid but semantically wrong for the manifest.
            restored.execute("DELETE FROM sample WHERE id=101")
            wrong_but_structural = restored.execute("PRAGMA integrity_check").fetchall()
            check("integrity_ok_does_not_prove_expected_content", wrong_but_structural == [("ok",)] and readback(restored) != expected())
            check("editing_sandbox_restore_does_not_edit_backup_or_source", readback(reader) == expected() and readback(source) == expected(changed=True))
            check("private_distinct_paths_delete_mode_and_no_open_write_transaction", len(set(p.resolve() for p in paths)) == 3 and all(p.resolve().parent == folder for p in paths) and journal_mode == "delete" and not source.in_transaction)
            output = {"python_version": platform.python_version(), "sqlite_version": sqlite3.sqlite_version,
                      "scope": "fresh private local files; sequential single thread; committed source before every backup; no WAL, concurrent writer, raw file copy, live database, process kill, power loss or media corruption test",
                      "units": {"ts_ms": "synthetic integer ms", "voltage_mv": "synthetic integer mV"},
                      "first_manifest": first_read, "later_source_and_second_manifest": expected(changed=True),
                      "backup_integrity_check": integrity, "readonly_write_error_code": readonly_code,
                      "structurally_valid_wrong_restore_integrity": wrong_but_structural,
                      "journal_mode": journal_mode, "existing_user_files_touched": False}
    finally:
        # Every connection above closes first. Delete only these verified exact names.
        for database in paths:
            for suffix in ("", "-journal", "-wal", "-shm"):
                target = database.with_name(database.name + suffix)
                if target.resolve().parent != folder or folder.parent != parent:
                    raise RuntimeError("cleanup path boundary")
                if target.exists():
                    target.unlink()
        folder.rmdir()  # Nonrecursive; unexpected files are not silently deleted.
    check("closed_private_fixture_cleaned_without_recursive_delete", not folder.exists())
    output["cleanup_completed"] = True
    output["checks"] = checks
    return output


if __name__ == "__main__":
    print(json.dumps(run_example(), ensure_ascii=False, indent=2))

بازیابیِ آزمایشی نیز انجام می‌شود: خوانندهٔ پشتیبان، محتوایش را با backup به یک اتصال :memory: تازه می‌برد. ردیف‌های آن محیط جدا دوباره با مرجع اولیه مقایسه می‌شوند. پایگاه دادهٔ مبدأ پاک یا جایگزین نمی‌شود و برای اثباتِ بازیابی، نیازی به خراب‌کردنِ آن نیست. اگر مقصدِ واقعیِ عملیاتی قرار بود بازنویسی شود، سیاست و شواهدِ جدا لازم داشت؛ این کد چنین عملیاتی ندارد.

همهٔ اتصال‌ها پیش از پاک‌سازی بسته می‌شوند. حذف فقط نام‌های دقیقِ سه فایل و هم‌نام‌های احتمالیِ journal یا WAL را، پس از بررسیِ ماندن در پوشهٔ اختصاصی، هدف می‌گیرد؛ حذف بازگشتی نداریم. پوشه فقط وقتی خالی است برداشته می‌شود. فایلِ ناشناخته بی‌صدا پاک نمی‌شود؛ پاک‌سازیِ این آزمایش، سیاست حذف نسخهٔ پشتیبان در کاربرد واقعی معرفی نشده است.

۴. سلامت ساختاری با محتوای درست یکسان نیست

PRAGMA integrity_check ساختار و سازگاری سطح پایین پایگاه داده را بررسی می‌کند؛ پاسخ ok یعنی در محدودهٔ آن بررسی خطایی پیدا نشده است. خودِ این دستور، خطاهای کلید خارجی را نمی‌یابد؛ foreign_key_check جدا اجرا می‌شود و در نمونهٔ ما خروجی خالی دارد. هیچ‌کدام نمی‌دانند نیاز کاربرد، دقیقاً سه نمونه با آن مقدارهای مشخص بوده است.

برای دیدن این تفاوت، در محیطِ بازیابی‌شدهٔ جدا، نمونهٔ ۱۰۱ را با DELETE عادی حذف می‌کنیم. integrity_check هنوز ok می‌دهد، ولی مقایسهٔ منطقی با مرجع شکست می‌خورد. فایل خراب نکرده‌ایم؛ یک پایگاه دادهٔ سالم با محتوای نامناسب برای معیارمان ساخته‌ایم. پشتیبانِ اصلی و مبدأ همچنان مستقل باقی می‌مانند. این شاهد از ادعای مبهمِ «فایل باز شد، پس بازیابی صحیح است» دقیق‌تر است.

{
  "python_version": "3.14.5",
  "sqlite_version": "3.50.4",
  "scope": "fresh private local files; sequential single thread; committed source before every backup; no WAL, concurrent writer, raw file copy, live database, process kill, power loss or media corruption test",
  "units": {
    "ts_ms": "synthetic integer ms",
    "voltage_mv": "synthetic integer mV"
  },
  "first_manifest": {
    "devices": [
      [
        1,
        "coil-A"
      ],
      [
        2,
        "rail-B"
      ]
    ],
    "samples": [
      [
        101,
        1,
        1000,
        1200
      ],
      [
        102,
        1,
        1010,
        null
      ],
      [
        201,
        2,
        1000,
        -50
      ]
    ],
    "metadata": [
      [
        "unit",
        "mV"
      ]
    ],
    "user_version": 7
  },
  "later_source_and_second_manifest": {
    "devices": [
      [
        1,
        "coil-A"
      ],
      [
        2,
        "rail-B"
      ]
    ],
    "samples": [
      [
        101,
        1,
        1000,
        1250
      ],
      [
        202,
        2,
        1020,
        300
      ]
    ],
    "metadata": [
      [
        "unit",
        "mV"
      ]
    ],
    "user_version": 7
  },
  "backup_integrity_check": [
    [
      "ok"
    ]
  ],
  "readonly_write_error_code": 8,
  "structurally_valid_wrong_restore_integrity": [
    [
      "ok"
    ]
  ],
  "journal_mode": "delete",
  "existing_user_files_touched": false,
  "cleanup_completed": true,
  "checks": {
    "committed_source_matches_hand_manifest": true,
    "closed_then_reopened_backup_matches_manifest": true,
    "tables_and_explicit_index_preserved": true,
    "metadata_signed_value_and_null_preserved": true,
    "backup_integrity_check_ok": true,
    "backup_foreign_key_check_separate_and_empty": true,
    "readonly_reopen_rejects_write": true,
    "later_source_changes_match_second_hand_manifest": true,
    "old_backup_unchanged_after_later_source_commit": true,
    "rollback_not_in_second_committed_manifest": true,
    "restore_into_fresh_private_memory_matches_old_manifest": true,
    "second_backup_captures_new_state_without_replacing_first": true,
    "integrity_ok_does_not_prove_expected_content": true,
    "editing_sandbox_restore_does_not_edit_backup_or_source": true,
    "private_distinct_paths_delete_mode_and_no_open_write_transaction": true,
    "closed_private_fixture_cleaned_without_recursive_delete": true
  }
}

در سناریوی نسخهٔ تازه، مبدأ مقدار ۱۰۱ را به ۱۲۵۰ تغییر می‌دهد، ۱۰۲ و ۲۰۱ را حذف و ۲۰۲ با مقدار ۳۰۰ اضافه می‌کند. پشتیبانِ قبلی هنوز سه ردیف آغازین را دارد؛ پشتیبانِ تازه فقط دو ردیفِ جدیدِ مورد انتظار را. یک UPDATE با مقدار ۹۹۹۹ نیز پیش از کپیِ دوم ROLLBACK می‌شود و وارد نسخهٔ ثبت‌شده نمی‌شود. نسخهٔ قدیمی، صرفاً به‌سبب تغییر مبدأ، تازه نشده است.

شانزده گروه واقعی با Python 3.14.5 و SQLite 3.50.4 اجرا شده‌اند؛ تعداد را با تکرارِ اجرا بزرگ نکرده‌ایم. مقایسهٔ مستقلِ ردیف‌ها، نسخه، NULL، اندیس، دو پشتیبان و بازیابیِ جدا ثبت است. اجرای ترتیبیِ موفق، خرابی رسانه، قطع برق، دوامِ متادیتا، ذخیرهٔ شبکه یا زمانِ بازیابی را آزمایش نمی‌کند. نسخهٔ قدیمیِ محیط نیز پیشنهادِ استقرار نیست؛ اینجا ارتقا یا آزمونِ خطاهای هم‌زمانی انجام نشده است.

کپی ثابتِ صفحه‌بندی حسگر برای پیمایشِ یک مجموعه ساخته شده بود و دیدِ خواندن در WAL عمرِ تراکنش را بررسی می‌کرد. اینجا معیارِ بازکردن و بازیابیِ فایلِ پشتیبان را آزمودیم. منابع در ۲۵ مهر ۱۴۰۵ خوانده شدند؛ بازنگریِ صفحهٔ backup نوامبر ۲۰۲۵، صفحهٔ PRAGMA اکتبر ۲۰۲۶ و مستندات Python 3.14.8 است؛ تاریخ‌ها از نسخهٔ اجرا جدا ثبت شده‌اند.

۵. چهار پرسش با پاسخ کامل

پرسش اول: فایلِ سالمِ دارای دو ردیف، پشتیبانِ صحیحِ نسخهٔ اولیه است؟ پاسخ: لزوماً خیر؛ معیارِ اولیه سه ردیفِ معلوم دارد. باید نسخه، عضویت و مقدارها با مرجع همان نسخه تطبیق داده شوند؛ ok ساختاری کافی نیست.

پرسش دوم: تغییرِ مبدأ، پشتیبانِ بسته‌شدهٔ قبلی را تازه می‌کند؟ پاسخ: در این آزمایش خیر؛ پشتیبان مستقل قبلاً کامل شده است. کپیِ تازه به مقصدِ دیگری دادهٔ جدید را می‌گیرد و هر دو نسخه جدا بررسی می‌شوند.

پرسش سوم: چرا نمونهٔ ۹۹۹۹ در کپیِ دوم نیست؟ پاسخ: UPDATE در تراکنشی انجام شد که پیش از backup با ROLLBACK پایان یافت. مرجعِ ثبت‌شده همان ۱۲۵۰ است؛ ادعایی دربارهٔ پشتیبان‌گیری وسطِ تراکنشِ نوشتن نداریم.

پرسش چهارم: این موفقیت، بازیابی پس از قطع برق را ثابت می‌کند؟ پاسخ: خیر؛ فقط کپی، بستن، بازکردن و بازیابیِ محلیِ مرتب اجرا شده‌اند. خطای رسانه، قطع فرایند و برق و معیار زمانِ بازیابی، شواهدِ جدا می‌خواهند و اینجا آزمایش نشده‌اند.

منابع و مطالعهٔ بیشتر

• SQLite؛ Online Backup API و محدودیت‌های کپی و بازیابی
• Python 3.14؛ Connection.backup و اتصال SQLite
• SQLite؛ integrity_check و foreign_key_check

بررسی منابع: ۲۵ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.