"""Observe SQLite WAL reader/checkpoint boundaries on disposable synthetic data.""" import argparse from contextlib import ExitStack from datetime import datetime, timezone import hashlib import json from pathlib import Path import platform import sqlite3 import subprocess import tempfile ROWS = 32 PAYLOAD_BYTES = 1024 BATCHES = 4 LIMITS = [ "One deterministic schedule on synthetic data, not a throughput benchmark or field sample.", "All SQLite calls run sequentially on one thread; read transactions overlap writes in lifetime only.", "No process crash, power loss, I/O fault, multi-writer race or backup procedure is tested.", "Exact frame counts and file sizes are observations of the recorded build, not universal constants.", "The installed SQLite version is recorded, not endorsed as patched or safe for production.", "PASSIVE checkpoint calls perform work; they are not passive telemetry reads.", ] def require(condition, message): if not condition: raise RuntimeError(message) def query(connection, statement, parameters=()): cursor = connection.execute(statement, parameters) try: return cursor.fetchall() finally: cursor.close() def expected(generation): return [(i, generation, bytes([65 + generation]) * PAYLOAD_BYTES) for i in range(ROWS)] def check_rows(connection, generation): rows = query(connection, "SELECT id, generation, payload FROM records ORDER BY id") require(rows == expected(generation), "Complete row oracle mismatch") return {"generation": generation, "rowCount": len(rows), "allRowsMatchOracle": True} def checkpoint(connection, path, mode, label): require(mode in ("PASSIVE", "RESTART", "TRUNCATE"), "Unsupported checkpoint mode") busy, log, done = query(connection, "PRAGMA main.wal_checkpoint(" + mode + ")")[0] wal = Path(str(path) + "-wal") return {"label": label, "mode": mode, "busy": busy, "logFrames": log, "checkpointedFrames": done, "walBytes": wal.stat().st_size if wal.exists() else 0, "walExists": wal.exists()} def run_case(root, pinned): name = "pinned-read-transaction" if pinned else "idle-open-connection" path = root / (name + ".db") require(not path.exists(), "Refusing existing database") with ExitStack() as stack: def connect(): connection = sqlite3.connect(path, timeout=0, isolation_level=None) stack.callback(connection.close) query(connection, "PRAGMA busy_timeout=0") query(connection, "PRAGMA synchronous=FULL") query(connection, "PRAGMA wal_autocheckpoint=0") return connection writer = connect() query(writer, "PRAGMA page_size=4096") require(query(writer, "PRAGMA journal_mode=WAL") == [("wal",)], "WAL unavailable") query(writer, "CREATE TABLE records(id INTEGER PRIMARY KEY, generation INTEGER, payload BLOB)") query(writer, "BEGIN IMMEDIATE") writer.executemany("INSERT INTO records VALUES(?,?,?)", expected(0)).close() query(writer, "COMMIT") require(checkpoint(writer, path, "TRUNCATE", "initial")["walBytes"] == 0, "Initial WAL not empty") # Seed a WAL-resident snapshot; a database-only reader has different lock semantics. query(writer, "BEGIN IMMEDIATE") query(writer, "UPDATE records SET generation=1, payload=?", (bytes([66]) * PAYLOAD_BYTES,)) query(writer, "COMMIT") reader = connect() if pinned: query(reader, "BEGIN") check_rows(reader, 1) require(reader.in_transaction == pinned, "Reader transaction precondition failed") settings = {key: query(writer, "PRAGMA " + key)[0][0] for key in ("page_size", "synchronous", "wal_autocheckpoint", "busy_timeout", "journal_mode")} observations = [] for generation in range(2, BATCHES + 2): query(writer, "BEGIN IMMEDIATE") query(writer, "UPDATE records SET generation=?, payload=?", (generation, bytes([65 + generation]) * PAYLOAD_BYTES)) query(writer, "COMMIT") observed = checkpoint(writer, path, "PASSIVE", "after-generation-" + str(generation)) observed["writerOracle"] = check_rows(writer, generation) observed["readerOracle"] = check_rows(reader, 1 if pinned else generation) observations.append(observed) blocked = checkpoint(writer, path, "TRUNCATE", "before-reader-release") require(blocked["busy"] == (1 if pinned else 0), "Unexpected TRUNCATE outcome") if pinned: query(reader, "ROLLBACK") require(not reader.in_transaction, "Read transaction still active") # Keep the connection open, separating connection lifetime from snapshot lifetime. restart = checkpoint(writer, path, "RESTART", "after-reader-release") truncate = checkpoint(writer, path, "TRUNCATE", "after-restart") final = check_rows(reader, BATCHES + 1) require(query(writer, "PRAGMA integrity_check") == [("ok",)], "Integrity check failed") return {"case": name, "settings": settings, "readerPinnedInitially": pinned, "observations": observations, "truncateBeforeRelease": blocked, "restartAfterRelease": restart, "truncateAfterRestart": truncate, "readerConnectionStillOpen": True, "finalOracle": final, "integrityCheck": "ok"} def run_observations(work_root=None): with tempfile.TemporaryDirectory(prefix="hesela-wal-readers-", dir=work_root) as temp: root = Path(temp) with sqlite3.connect(":memory:") as connection: source_id = query(connection, "SELECT sqlite_source_id()")[0][0] compile_options = [row[0] for row in query(connection, "PRAGMA compile_options")] try: fs = subprocess.check_output(["stat", "-f", "-c", "%T", str(root)], text=True).strip() except (OSError, subprocess.CalledProcessError): fs = "unavailable" return { "schemaVersion": 1, "evidenceType": "measured-synthetic-workload", "generatedAt": datetime.now(timezone.utc).isoformat(), "scriptSha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "environment": {"sqlite": sqlite3.sqlite_version, "sqliteSourceId": source_id, "sqliteCompileOptions": compile_options, "python": platform.python_version(), "os": platform.system(), "kernel": platform.release(), "statfsLabel": fs}, "workload": {"rows": ROWS, "payloadBytes": PAYLOAD_BYTES, "updateBatches": BATCHES, "initialWalGeneration": 1, "lastGeneration": BATCHES + 1, "randomness": "none"}, "caseCount": 2, "cases": [run_case(root, True), run_case(root, False)], "limits": LIMITS, "sources": [ "https://www.sqlite.org/wal.html", "https://www.sqlite.org/c3ref/wal_checkpoint_v2.html", "https://www.sqlite.org/pragma.html#pragma_wal_checkpoint", "https://www.sqlite.org/isolation.html", ], } def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--work-root", type=Path, required=True, help="Existing disposable-storage directory") parser.add_argument("--output", type=Path, required=True, help="New JSON filename; never overwritten") args = parser.parse_args() if not args.work_root.is_dir(): parser.error("--work-root must be an existing directory") if args.output.exists() or args.output.is_symlink(): parser.error("--output must not already exist") result = run_observations(args.work_root) with args.output.open("x", encoding="utf-8") as output: json.dump(result, output, indent=2) output.write("\n") if __name__ == "__main__": main()