#!/usr/bin/env python3 """Original synthetic SQLite pagination exercise; Python 3.11+, standard library. Run: python3.11 fde-pagination-consistency.py all --page-size 2 Each case creates and deletes its own temporary SQLite WAL database. There are no network requests, credentials, inputs from other files, or durable writes. These observations concern this ordered integer-ID dataset and SQLite's WAL transactions. They are not production benchmarks or portable isolation promises. A keyset/high-water bound is not a snapshot; a snapshot is not the latest data. A pagination cursor is not authorization: this lab implements neither API tokens nor access control. All mutations commit immediately after the first page. """ import argparse from collections import Counter import json from pathlib import Path import sqlite3 import sys from tempfile import TemporaryDirectory CASES = ("offset-insert", "offset-delete", "keyset-insert", "high-water", "snapshot") INITIAL_IDS = (10, 20, 30, 40) MAX_PAGES = 8 # At most six rows, page size >= 1, plus the terminal empty page. def ids(connection): return [row[0] for row in connection.execute("SELECT id FROM items ORDER BY id")] def mutations(case): if case == "offset-insert": return [("INSERT INTO items (id) VALUES (?)", [5])] if case == "offset-delete": return [("DELETE FROM items WHERE id = ?", [10])] statements = [("INSERT INTO items (id) VALUES (?)", [15]), ("INSERT INTO items (id) VALUES (?)", [50])] if case == "snapshot": statements.append(("DELETE FROM items WHERE id = ?", [30])) return statements def run_case(case, page_size): reader = writer = None report = None with TemporaryDirectory(prefix="pagination-lab-") as scratch: database = Path(scratch) / "synthetic.sqlite3" try: # Autocommit lets each ordinary SELECT see the latest committed state. writer = sqlite3.connect(database, isolation_level=None, timeout=0) journal_mode = writer.execute("PRAGMA journal_mode=WAL").fetchone()[0] if journal_mode != "wal": raise RuntimeError("This exercise requires SQLite WAL mode") writer.execute("CREATE TABLE items (id INTEGER PRIMARY KEY)") writer.executemany("INSERT INTO items (id) VALUES (?)", [(i,) for i in INITIAL_IDS]) initial_ids = ids(writer) high_water_id = max(initial_ids) if case == "high-water" else None reader = sqlite3.connect(database, isolation_level=None, timeout=0) snapshot = case == "snapshot" if snapshot: reader.execute("BEGIN") # The first page SELECT below establishes the WAL read snapshot. pages, received = [], [] last_id, offset = 0, 0 mutation = None for page_number in range(1, MAX_PAGES + 1): if case in ("keyset-insert", "high-water"): sql = "SELECT id FROM items WHERE id > ?" parameters = [last_id] if high_water_id is not None: sql += " AND id <= ?" parameters.append(high_water_id) sql += " ORDER BY id LIMIT ?" parameters.append(page_size) else: sql = "SELECT id FROM items ORDER BY id LIMIT ? OFFSET ?" parameters = [page_size, offset] page_ids = [row[0] for row in reader.execute(sql, parameters).fetchall()] pages.append({"sql": sql, "parameters": parameters, "ids": page_ids, "reader_in_transaction": reader.in_transaction}) received.extend(page_ids) if page_number == 1: statements = mutations(case) writer.execute("BEGIN") for statement, values in statements: writer.execute(statement, values) writer.execute("COMMIT") mutation = {"after_page": 1, "statements": [{"sql": s, "parameters": p} for s, p in statements], "committed": not writer.in_transaction, "after_ids": ids(writer)} if not page_ids: break last_id = page_ids[-1] offset += page_size else: raise RuntimeError("Pagination exceeded the exercise's fixed page bound") final_ids = ids(writer) if snapshot: reader.execute("COMMIT") fresh_ids = ids(reader) # After ending the snapshot, this reader sees the committed changes. counts = Counter(received) report = { "case": case, "transaction_mode": "BEGIN; first SELECT establishes snapshot; COMMIT after empty page" if snapshot else "autocommit; each page is a separate read transaction", "journal_mode": journal_mode, "separate_reader_writer_connections": reader is not writer, "initial_ids": initial_ids, "high_water_id": high_water_id, "pages": pages, "mutation": mutation, "received_ids": received, "duplicate_ids": sorted(i for i, count in counts.items() if count > 1), "missing_from_initial_ids": sorted(set(initial_ids) - set(received)), "newly_read_ids": sorted(set(received) - set(initial_ids)), "final_current_ids_missed": sorted(set(final_ids) - set(received)), "final_current_ids": final_ids, "fresh_reader_ids": fresh_ids, } finally: if reader is not None: reader.close() if writer is not None: writer.close() report["connections_closed"] = True report["temporary_database_removed"] = not database.exists() return report def page_size_argument(value): try: size = int(value) except ValueError: raise argparse.ArgumentTypeError("page size must be an integer from 1 to 4") from None if not 1 <= size <= 4: raise argparse.ArgumentTypeError("page size must be an integer from 1 to 4") return size def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("case", choices=("all", *CASES), nargs="?", default="all") parser.add_argument("--page-size", type=page_size_argument, default=2, help="rows per page, 1..4 (default: 2); mutations follow the first page") arguments = parser.parse_args() if sys.version_info < (3, 11): parser.error("Python 3.11 or newer is required") selected = CASES if arguments.case == "all" else (arguments.case,) print(json.dumps({ "scope": "Original synthetic local SQLite WAL exercise. Creates and removes temporary databases; no network, authorization, or production API. Cursors do not grant authorization. High-water is not a snapshot; a snapshot is not the latest data.", "page_size": arguments.page_size, "cases": [run_case(case, arguments.page_size) for case in selected], }, indent=2)) if __name__ == "__main__": main()