Запоздавший INSERT при откате: разбор сбоя на стенде SQLite

from contextlib import closing
from pathlib import Path
import hashlib
import json
import os
import platform
import sqlite3
import subprocess
import sys
import tempfile


RELEASES = {"legacy", "native_v2", "bridge"}
MAX_DELAY_MS = 86_400_000


class ContractError(ValueError):
    pass


def require(condition, message):
    if not condition:
        raise ContractError(message)


def integer(value, maximum):
    require(type(value) is int and 0 <= value <= maximum,
            "delay must be an integer within range")
    return value


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        require(key not in result, "duplicate JSON key")
        result[key] = value
    return result


def decode(payload, release):
    require(release in RELEASES, "unknown release")
    try:
        item = json.loads(payload, object_pairs_hook=unique_object)
    except (json.JSONDecodeError, TypeError) as error:
        raise ContractError("invalid JSON") from error
    require(type(item) is dict and set(item) == {"schema", "delay"},
            "unexpected payload fields")
    version = item["schema"]
    require(type(version) is int and version in (1, 2),
            "unsupported schema")
    if version == 1:
        return integer(item["delay"], MAX_DELAY_MS // 1000) * 1000
    require(release != "legacy", "legacy cannot read schema 2")
    delay = item["delay"]
    require(type(delay) is dict and set(delay) == {"unit", "value"},
            "unexpected delay fields")
    require(delay["unit"] == "ms", "unsupported delay unit")
    return integer(delay["value"], MAX_DELAY_MS)


def encode(delay_ms, release):
    require(release in RELEASES, "unknown release")
    integer(delay_ms, MAX_DELAY_MS)
    if release == "native_v2":
        item = {"schema": 2, "delay": {"unit": "ms", "value": delay_ms}}
    else:
        require(delay_ms % 1000 == 0, "v1 cannot represent subsecond delay")
        item = {"schema": 1, "delay": delay_ms // 1000}
    return json.dumps(item, sort_keys=True, separators=(",", ":"))


def connect(path, mode):
    uri = Path(path).resolve().as_uri() + "?mode=" + mode
    return sqlite3.connect(uri, uri=True,
                           autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL)


def worker(path, release, operation, value=None):
    require(release in RELEASES, "unknown release")
    require(operation in {"write", "read"}, "unknown operation")
    with closing(connect(path, "rw" if operation == "write" else "ro")) as db:
        if operation == "write":
            payload = encode(value, release)
            with db:
                row_id = db.execute("INSERT INTO jobs(payload) VALUES (?)",
                                    (payload,)).lastrowid
            return {"row": row_id, "schema": json.loads(payload)["schema"],
                    "delay_ms": value}
        rows = db.execute("SELECT id, payload FROM jobs ORDER BY id").fetchall()
        delays = []
        for row_id, payload in rows:
            try:
                delays.append(decode(payload, release))
            except ContractError as error:
                raise ContractError(f"job {row_id}: {error}") from error
        return {"delays_ms": delays}


def snapshot(path):
    with closing(connect(path, "ro")) as db:
        rows = db.execute(
            "SELECT id, CAST(payload AS BLOB) FROM jobs ORDER BY id"
        ).fetchall()
        integrity = db.execute("PRAGMA integrity_check").fetchall()
    return {"rows": [{"id": row_id, "hex": raw.hex(),
                      "sha256": hashlib.sha256(raw).hexdigest()}
                     for row_id, raw in rows], "integrity": integrity}


def run_demo(extra_checks=None):
    root = Path(tempfile.mkdtemp(prefix="rollback-run-", dir=".")).resolve()
    report = {"python": sys.version, "sqlite": sqlite3.sqlite_version,
              "platform": platform.platform(), "root": str(root),
              "script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
              "checks": [] if extra_checks is None else extra_checks,
              "check_scope": {"scenario_invariants": True,
                              "additional_checks_supplied": extra_checks is not None},
              "scenarios": {}}

    def scenario(name):
        path = root / (name + ".db")
        with closing(connect(path, "rwc")) as db:
            with db:
                db.execute("CREATE TABLE jobs(id INTEGER PRIMARY KEY, payload TEXT NOT NULL)")
        data = {"trace": []}
        report["scenarios"][name] = data
        print(name.upper())

        def step(release, operation, value=None, expected_error=None):
            command = [sys.executable, str(Path(__file__).resolve()),
                       "worker", str(path), release, operation]
            if value is not None:
                command.append(str(value))
            child = subprocess.run(command, capture_output=True, text=True,
                                   encoding="utf-8", check=False)
            require(not child.stderr, "worker wrote to stderr")
            result = json.loads(child.stdout)
            data["trace"].append({"release": release, "operation": operation,
                                  "exit": child.returncode, "result": result})
            if expected_error is not None:
                require(child.returncode == 2 and result.get("error") == expected_error,
                        "unexpected worker error")
                print(f"{release} {operation}: ERROR {result['error']}")
            else:
                require(child.returncode == 0 and result.get("ok") is True,
                        "worker did not succeed")
                if operation == "write":
                    print(f"{release} write: row={result['row']} schema={result['schema']} delay_ms={value}")
                else:
                    print(f"{release} read: OK delays_ms={result['delays_ms']}")
            return result

        return path, data, step

    path, data, step = scenario("unsafe")
    step("legacy", "write", 2000)
    step("legacy", "read")
    step("native_v2", "read")
    step("native_v2", "write", 1500)
    before = snapshot(path)
    step("legacy", "read", expected_error="job 2: legacy cannot read schema 2")
    require(step("bridge", "read")["delays_ms"] == [2000, 1500], "wrong recovery")
    after = snapshot(path)
    require(before == after, "payloads changed during rollback or bridge read")
    require(after["integrity"] == [("ok",)], "database integrity failed")
    data.update({"before": before, "after": after, "payloads_unchanged": True})
    print("payloads unchanged: True; database integrity: ok")

    path, data, step = scenario("safe")
    step("legacy", "write", 2000)
    step("bridge", "read")
    step("bridge", "write", 3000)
    with closing(connect(path, "ro")) as db:
        versions = [json.loads(row[0])["schema"] for row in db.execute("SELECT payload FROM jobs")]
    require(versions == [1, 1], "bridge wrote a new format too early")
    step("legacy", "read")
    print("compatible reader phase complete; enable v2 writer")
    step("native_v2", "write", 1500)
    before = snapshot(path)
    require(step("bridge", "read")["delays_ms"] == [2000, 3000, 1500], "wrong safe rollback")
    step("legacy", "read", expected_error="job 3: legacy cannot read schema 2")
    after = snapshot(path)
    require(before == after, "payloads changed during safe rollback")
    require(after["integrity"] == [("ok",)], "database integrity failed")
    data.update({"bridge_phase_schemas": versions, "before": before,
                 "after": after, "payloads_unchanged": True})
    print("payloads unchanged: True; rollback floor: bridge")
    with (root / "qa.json").open("x", encoding="utf-8") as stream:
        json.dump(report, stream, indent=2, ensure_ascii=False)
    return root, report


if __name__ == "__main__":
    if len(sys.argv) == 1:
        run_demo()
    else:
        try:
            require(len(sys.argv) in (5, 6) and sys.argv[1] == "worker",
                    "invalid arguments: worker DATABASE RELEASE read|write [DELAY_MS]")
            operation = sys.argv[4]
            require((operation == "read" and len(sys.argv) == 5) or
                    (operation == "write" and len(sys.argv) == 6),
                    "read takes no delay; write requires one integer delay in milliseconds")
            try:
                value = int(sys.argv[5]) if operation == "write" else None
            except ValueError as error:
                raise ContractError("delay must be an integer in milliseconds") from error
            result = worker(sys.argv[2], sys.argv[3], sys.argv[4],
                            value)
        except (ContractError, sqlite3.Error) as error:
            print(json.dumps({"ok": False, "pid": os.getpid(), "error": str(error)}))
            raise SystemExit(2)
        print(json.dumps({"ok": True, "pid": os.getpid(), **result}))

"""Python >= 3.12; stdlib; bounded IPC waits, no timing-based scheduling.

All production write paths must enforce the fence. SQLite serializes writers
to this one database; this is not a distributed rollout protocol.
"""
from contextlib import closing
from pathlib import Path
import hashlib
import json
import multiprocessing as mp
import os
import platform
import sqlite3
import sys
import tempfile

import rollback_lab as lab

HERE = Path(__file__).resolve().parent
OLD = ("rollback_lab.py",)
WAIT = 20  # Failure bound, not an ordering mechanism.
require = lab.require


def hashes(names):
    return {name: hashlib.sha256((HERE / name).read_bytes()).hexdigest()
            for name in names}


def policy(db):
    epoch, cap = db.execute("SELECT epoch, cap FROM policy WHERE id=1").fetchone()
    return {"epoch": epoch, "cap": cap}


def initialize(path):
    with closing(lab.connect(path, "rwc")) as db, db:
        db.execute("CREATE TABLE jobs(id INTEGER PRIMARY KEY, payload TEXT NOT NULL)")
        db.execute("CREATE TABLE policy(id INTEGER PRIMARY KEY CHECK(id=1), "
                   "epoch INTEGER NOT NULL, cap INTEGER NOT NULL CHECK(cap IN (1,2)))")
        db.execute("INSERT INTO policy VALUES(1,7,2)")
        # Fixture setup, before either worker exists.
        db.execute("INSERT INTO jobs(payload) VALUES (?)", (lab.encode(2000, "legacy"),))


def inspect(path):
    with closing(lab.connect(path, "ro")) as db, db:
        db.execute("BEGIN")
        current = policy(db)
        rows = db.execute("SELECT id, CAST(payload AS BLOB) FROM jobs ORDER BY id").fetchall()
        integrity = db.execute("PRAGMA integrity_check").fetchall()
    return {"policy": current, "row_count": len(rows), "integrity": integrity,
            "rows": [{"id": row_id, "hex": raw.hex(),
                      "sha256": hashlib.sha256(raw).hexdigest()} for row_id, raw in rows]}


def fenced_insert(path, ticket, payload):
    lab.decode(payload, "bridge")  # Validate bytes before using their schema.
    schema = json.loads(payload)["schema"]
    with closing(lab.connect(path, "rw")) as db, db:
        db.execute("BEGIN IMMEDIATE")
        current = policy(db)
        require(ticket == current["epoch"],
                f"stale epoch: ticket={ticket} current={current['epoch']}")
        require(schema <= current["cap"],
                f"schema {schema} exceeds write cap {current['cap']}")
        row_id = db.execute("INSERT INTO jobs(payload) VALUES (?)", (payload,)).lastrowid
    return {"row": row_id, "schema": schema, "checked_policy": current}


def gate(path, fenced=True):
    with closing(lab.connect(path, "rw")) as db, db:
        db.execute("BEGIN IMMEDIATE")
        db.execute("UPDATE policy SET epoch=epoch+?, cap=1 WHERE id=1", (int(fenced),))
        current, count_v2, errors = policy(db), 0, []
        for row_id, payload in db.execute("SELECT id, payload FROM jobs ORDER BY id"):
            try:
                item = json.loads(payload)
                count_v2 += int(type(item) is dict and item.get("schema") == 2)
            except (json.JSONDecodeError, TypeError):
                pass
            try:
                lab.decode(payload, "legacy")
            except lab.ContractError as error:
                errors.append(f"job {row_id}: {error}")
        admitted = not errors if fenced else count_v2 == 0
        # Commit the write restriction even when old-reader admission is denied.
    return {"admitted": admitted, "policy": current, "count_v2": count_v2,
            "incompatible": errors, "validated_all_payloads": fenced}


def child(path, pipe, release, fenced):
    try:
        if release:
            with closing(lab.connect(path, "ro")) as db:
                ticket = policy(db)["epoch"]
            payload = lab.encode(1500 if release == "native_v2" else 3000, release)
            pipe.send({"event": "parked", "pid": os.getpid(), "ticket": ticket,
                       "release": release, "payload_hex": payload.encode().hex()})
            require(pipe.recv() == "commit", "writer expected commit")
            try:
                if fenced:
                    result = fenced_insert(path, ticket, payload)
                else:  # Intentionally unsafe: an in-flight write ignores policy.
                    with closing(lab.connect(path, "rw")) as db, db:
                        row_id = db.execute("INSERT INTO jobs(payload) VALUES (?)",
                                            (payload,)).lastrowid
                    result = {"row": row_id, "schema": json.loads(payload)["schema"]}
                pipe.send({"event": "committed", "pid": os.getpid(), **result})
            except lab.ContractError as error:
                pipe.send({"event": "rejected", "pid": os.getpid(), "error": str(error)})
            require(pipe.recv() == "stop", "writer expected stop")
        else:
            pipe.send({"event": "reader_ready", "pid": os.getpid()})
            while (command := pipe.recv()) != "stop":
                require(command == "read", "reader expected read")
                try:
                    result = {"ok": True, **lab.worker(path, "legacy", "read")}
                except lab.ContractError as error:
                    result = {"ok": False, "error": str(error)}
                pipe.send({"event": "read", "pid": os.getpid(), **result})
    except Exception as error:
        pipe.send({"event": "fatal", "pid": os.getpid(), "error": repr(error)})
    finally:
        pipe.close()


class Peer:
    def __init__(self, path, release=None, fenced=True):
        context = mp.get_context("spawn")
        self.pipe, other = context.Pipe()
        self.process = context.Process(target=child, args=(str(path), other, release, fenced))
        self.process.start()
        other.close()
        self.ready = self.receive()

    def receive(self):
        require(self.pipe.poll(WAIT), "IPC timeout")
        message = self.pipe.recv()
        require(message["event"] != "fatal", f"child failed: {message}")
        require(message["pid"] == self.process.pid, "unexpected child PID")
        return message

    def request(self, command):
        self.pipe.send(command)
        return self.receive()

    def close(self):
        try:
            if self.process.is_alive():
                self.pipe.send("stop")
            self.process.join(WAIT)
            require(not self.process.is_alive(), "child did not stop")
            require(self.process.exitcode == 0, "child exited unsuccessfully")
        finally:
            if self.process.is_alive():
                self.process.terminate()
                self.process.join(WAIT)
            self.pipe.close()


def run_demo():
    require(sys.version_info >= (3, 12), "Python 3.12 or newer is required")
    root = Path(tempfile.mkdtemp(prefix="race-run-", dir=HERE))
    original = hashes(OLD)
    report = {"python": sys.version, "sqlite": sqlite3.sqlite_version,
              "platform": platform.platform(), "parent_pid": os.getpid(), "root": str(root),
              "script_sha256": hashes((*OLD, "rollback_race.py", *(
                  name for name in ("test_rollback_race.py",) if (HERE / name).exists()))),
              "ordering": "Pipe acknowledgments and BEGIN IMMEDIATE; no sleeps",
              "scenarios": {}}
    lines = []

    def log(line):
        lines.append(line)
        print(line, flush=True)

    for name in ("UNSAFE_GATE", "FENCED_GATE", "RESIDUAL_V2", "WRITER_FIRST"):
        path, trace, peers = root / (name + ".db"), [], []
        initialize(path)
        data = {"initial": inspect(path), "trace": trace}
        report["scenarios"][name] = data

        def event(label, **facts):
            trace.append({"seq": len(trace) + 1, "event": label, **facts})
            log(f"{name} {len(trace):02d} {label}: "
                + json.dumps(facts, sort_keys=True, ensure_ascii=False))

        def peer(release=None, fenced=True):
            item = Peer(path, release, fenced)
            peers.append(item)
            return item

        try:
            if name == "RESIDUAL_V2":
                result = fenced_insert(path, 7, lab.encode(1500, "native_v2"))
                event("residual_committed", **result)
                before = inspect(path)
                decision = gate(path)
                event("gate_committed", **decision)
                data["after_gate"] = inspect(path)
                require(not decision["admitted"], "gate admitted residual v2")
                require(decision["policy"] == {"epoch": 8, "cap": 1}, "restriction lost")
                bridge = lab.worker(path, "bridge", "read")
                event("bridge_read", **bridge)
                require(bridge["delays_ms"] == [2000, 1500], "bridge lost precision")
                require(before["rows"] == inspect(path)["rows"], "gate changed payloads")
            else:
                writer = peer("native_v2", fenced=name != "UNSAFE_GATE")
                require(writer.ready["event"] == "parked", "writer not parked")
                require(writer.ready["ticket"] == 7, "wrong prepared ticket")
                event("writer_parked_before_insert", **{k: v for k, v in writer.ready.items()
                                                       if k != "event"})
                if name == "WRITER_FIRST":
                    result = writer.request("commit")
                    require(result["event"] == "committed", "writer-first failed")
                    event("writer_committed_before_gate", **{k: v for k, v in result.items()
                                                            if k != "event"})
                decision = gate(path, fenced=name != "UNSAFE_GATE")
                event("gate_committed", **decision)
                data["after_gate"] = inspect(path)
                if name == "WRITER_FIRST":
                    require(not decision["admitted"] and decision["count_v2"] == 1,
                            "gate missed an earlier commit")
                else:
                    require(decision["admitted"] and decision["count_v2"] == 0,
                            "unexpected initial gate refusal")
                    reader = peer()
                    event("legacy_reader_started", pid=reader.ready["pid"])
                    first = reader.request("read")
                    require(first.get("delays_ms") == [2000], "initial legacy read failed")
                    event("legacy_read_before_release", result=first)
                    live = {"writer_pid": writer.process.pid, "reader_pid": reader.process.pid,
                            "writer_alive": writer.process.is_alive(),
                            "reader_alive": reader.process.is_alive()}
                    require(live["writer_alive"] and live["reader_alive"], "no process overlap")
                    require(len({os.getpid(), live["writer_pid"], live["reader_pid"]}) == 3,
                            "workers are not distinct processes")
                    event("both_processes_alive_before_release", **live)
                    data["before_release"] = inspect(path)
                    event("release_writer")
                    result = writer.request("commit")
                    event("writer_result", result=result)
                    data["after_writer"] = inspect(path)
                    require(result["event"] == ("committed" if name == "UNSAFE_GATE"
                                                else "rejected"), "wrong write outcome")
                    if name == "FENCED_GATE":
                        require(result["error"] == "stale epoch: ticket=7 current=8",
                                "wrong fence rejection")
                        require(inspect(path)["rows"] == data["initial"]["rows"],
                                "rejected writer inserted a row")
                    after = reader.request("read")
                    event("legacy_read_after_writer", result=after)
                    require(after["ok"] == (name == "FENCED_GATE"), "wrong reader outcome")
                    if name == "UNSAFE_GATE":
                        require(after["error"] == "job 2: legacy cannot read schema 2",
                                "unexpected legacy failure")
                    else:
                        bridge = peer("bridge")
                        event("bridge_writer_parked", pid=bridge.ready["pid"],
                              ticket=bridge.ready["ticket"])
                        require(bridge.ready["ticket"] == 8, "bridge did not refresh its ticket")
                        result = bridge.request("commit")
                        require(result["event"] == "committed" and result["schema"] == 1,
                                "bridge did not commit v1")
                        event("bridge_writer_committed", result=result)
                        final_read = reader.request("read")
                        require(final_read.get("delays_ms") == [2000, 3000],
                                "legacy cannot read fresh bridge write")
                        event("legacy_read_after_bridge", result=final_read)
            data["gate"] = decision
            data["final"] = inspect(path)
            require(data["final"]["rows"][0] == data["initial"]["rows"][0], "seed changed")
            require(data["final"]["integrity"] == [("ok",)], "database integrity failed")
            # Repeated reads must preserve every row ID and payload byte.
            lab.worker(path, "bridge", "read")
            require(inspect(path) == data["final"], "read changed data or policy")
        finally:
            for item in reversed(peers):
                item.close()
        data["children_exitcodes"] = [item.process.exitcode for item in peers]
    require(hashes(OLD) == original, "an original script changed")
    report["original_scripts_unchanged"] = True
    report["original_scripts_checked"] = list(OLD)
    with (root / "qa.json").open("x", encoding="utf-8") as stream:
        json.dump(report, stream, indent=2, ensure_ascii=False)
    log(f"PASS 4 scenarios; artifacts: {root}")
    (root / "stdout.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
    return root, report


if __name__ == "__main__":
    run_demo()

from pathlib import Path
import contextlib
import io
import json
import sqlite3
import subprocess
import sys
import tempfile

import rollback_lab as lab


checks = []


def good(name, operation, expected):
    actual = operation()
    if actual != expected:
        raise RuntimeError(f"{name}: {actual!r} != {expected!r}")
    checks.append({"name": name, "ok": True})


def bad(name, operation, message):
    try:
        operation()
    except lab.ContractError as error:
        if message not in str(error):
            raise RuntimeError(f"{name}: unexpected error {error}") from error
        checks.append({"name": name, "ok": True, "error": str(error)})
    else:
        raise RuntimeError(f"{name}: invalid input accepted")


V1 = '{"schema":1,"delay":2}'
V2 = '{"schema":2,"delay":{"unit":"ms","value":1500}}'
for reader in ("legacy", "native_v2", "bridge"):
    good(f"matrix:{reader}:v1", lambda r=reader: lab.decode(V1, r), 2000)
    if reader == "legacy":
        bad("matrix:legacy:v2", lambda: lab.decode(V2, "legacy"), "cannot read")
    else:
        good(f"matrix:{reader}:v2", lambda r=reader: lab.decode(V2, r), 1500)

for reader in ("legacy", "native_v2", "bridge"):
    for seconds in (0, 1, 86400):
        good(f"v1-boundary:{reader}:{seconds}",
             lambda r=reader, s=seconds: lab.decode(json.dumps({"schema": 1, "delay": s}), r),
             seconds * 1000)
    for value in (True, -1, 86401, 1.5, "2", None):
        bad(f"v1-invalid:{reader}:{value!r}",
            lambda r=reader, v=value: lab.decode(json.dumps({"schema": 1, "delay": v}), r),
            "integer within range")

for reader in ("native_v2", "bridge"):
    for ms in (0, 1, 1500, 86_400_000):
        payload = json.dumps({"schema": 2, "delay": {"unit": "ms", "value": ms}})
        good(f"v2-boundary:{reader}:{ms}", lambda r=reader, p=payload: lab.decode(p, r), ms)
    for value in (True, -1, 86_400_001, 1.5, "2", None):
        bad(f"v2-invalid:{reader}:{value!r}",
            lambda r=reader, v=value: lab.decode(json.dumps({"schema": 2, "delay": {"unit": "ms", "value": v}}), r),
            "integer within range")

malformed = [
    ('{"schema":2,"delay":{"unit":"ms","value":"1500"}}', "integer within range"),
    ('{"schema":2,"delay":{"unit":"s","value":2}}', "unsupported delay unit"),
    ('{"schema":2,"delay":{"unit":true,"value":2}}', "unsupported delay unit"),
    ('{"schema":2,"delay":{"value":2}}', "unexpected delay fields"),
    ('{"schema":2,"delay":{"unit":"ms","value":2,"extra":0}}', "unexpected delay fields"),
    ('{"schema":3,"delay":2}', "unsupported schema"),
    ('{"schema":true,"delay":2}', "unsupported schema"),
    ('{"schema":1.0,"delay":2}', "unsupported schema"),
    ('{"schema":1,"delay":2,"extra":0}', "unexpected payload fields"),
    ('{"schema":1}', "unexpected payload fields"),
    ('[]', "unexpected payload fields"),
    ('null', "unexpected payload fields"),
    ('{"schema":1,"delay":2,"delay":3}', "duplicate JSON key"),
    ('{"schema":2,"delay":{"unit":"ms","unit":"s","value":2}}', "duplicate JSON key"),
    ('{"schema":', "invalid JSON"),
]
for i, (payload, error) in enumerate(malformed):
    bad(f"malformed:{i}", lambda p=payload: lab.decode(p, "bridge"), error)

for writer in ("legacy", "native_v2", "bridge"):
    expected = 2 if writer == "native_v2" else 1
    good(f"writer-format:{writer}", lambda r=writer: json.loads(lab.encode(2000, r))["schema"], expected)
    for ms in (0, 1000, 86_400_000):
        good(f"writer-roundtrip:{writer}:{ms}", lambda r=writer, v=ms: lab.decode(lab.encode(v, r), r), ms)
    for ms in (True, -1, 86_400_001, 1.0, "1000", None):
        bad(f"writer-invalid:{writer}:{ms!r}", lambda r=writer, v=ms: lab.encode(v, r), "integer within range")
    if writer != "native_v2":
        bad(f"no-rounding:{writer}", lambda r=writer: lab.encode(1500, r), "cannot represent")

bad("unknown-reader", lambda: lab.decode(V1, "v3"), "unknown release")
bad("unknown-writer", lambda: lab.encode(1000, "v3"), "unknown release")

# Neither read-only nor read/write opening may fabricate a missing database.
negative_root = Path(tempfile.mkdtemp(prefix="rollback-negative-", dir=".")).resolve()
for operation in ("read", "write"):
    missing = negative_root / ("missing-" + operation + ".db")
    try:
        lab.worker(missing, "bridge", operation, 2000 if operation == "write" else None)
    except sqlite3.OperationalError:
        good(f"missing-file-not-created:{operation}", lambda: missing.exists(), False)
    else:
        raise RuntimeError("missing database was accepted")

cli_path = negative_root / "cli-missing.db"
cli_cases = [
    ("bad-command", ["unknown"], "invalid arguments"),
    ("missing-delay", ["worker", str(cli_path), "bridge", "write"], "write requires"),
    ("extra-read-delay", ["worker", str(cli_path), "bridge", "read", "2000"], "read takes no delay"),
    ("noninteger-delay", ["worker", str(cli_path), "bridge", "write", "nope"], "delay must be an integer"),
    ("missing-database", ["worker", str(cli_path), "bridge", "write", "2000"], "unable to open database file"),
]
for name, arguments, message in cli_cases:
    child = subprocess.run([sys.executable, str(Path(lab.__file__).resolve()), *arguments],
                           capture_output=True, text=True, encoding="utf-8", check=False)
    good(f"cli:{name}:exit", lambda c=child: c.returncode, 2)
    good(f"cli:{name}:stderr", lambda c=child: c.stderr, "")
    result = json.loads(child.stdout)
    good(f"cli:{name}:ok", lambda r=result: r["ok"], False)
    good(f"cli:{name}:message", lambda r=result, m=message: m in r["error"], True)
    good(f"cli:{name}:database-not-created", lambda: cli_path.exists(), False)

rejected_write_db = negative_root / "rejected-write.db"
with contextlib.closing(lab.connect(rejected_write_db, "rwc")) as db:
    with db:
        db.execute("CREATE TABLE jobs(id INTEGER PRIMARY KEY, payload TEXT NOT NULL)")
lab.worker(rejected_write_db, "legacy", "write", 2000)
before_reject = lab.snapshot(rejected_write_db)
bad("rejected-bridge-write", lambda: lab.worker(rejected_write_db, "bridge", "write", 1500),
    "cannot represent")
good("rejected-write-no-change", lambda: lab.snapshot(rejected_write_db), before_reject)

captured = io.StringIO()
with contextlib.redirect_stdout(captured):
    root, report = lab.run_demo(extra_checks=checks)
with (root / "stdout.txt").open("x", encoding="utf-8", newline="\n") as stream:
    stream.write(captured.getvalue())
with (root / "checks.json").open("x", encoding="utf-8") as stream:
    json.dump({"passed": len(checks), "checks": checks}, stream, ensure_ascii=False, indent=2)
print(captured.getvalue(), end="")
print("checks passed:", len(checks))
print("qa directory:", root)

"""Behavioral checks; every temporary database is retained beside the scripts."""
from contextlib import closing, redirect_stdout
from io import StringIO
from pathlib import Path
import json
import tempfile
import unittest

import rollback_lab as lab
import rollback_race as race


class ProcessScenarios(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        with redirect_stdout(StringIO()):
            cls.root, cls.report = race.run_demo()

    def scenario(self, name):
        return self.report["scenarios"][name]

    def test_unsafe_order_and_live_distinct_processes(self):
        data = self.scenario("UNSAFE_GATE")
        trace = data["trace"]
        self.assertEqual([item["event"] for item in trace], [
            "writer_parked_before_insert", "gate_committed", "legacy_reader_started",
            "legacy_read_before_release", "both_processes_alive_before_release",
            "release_writer", "writer_result", "legacy_read_after_writer"])
        live = trace[4]
        self.assertTrue(live["writer_alive"] and live["reader_alive"])
        self.assertEqual(len({live["reader_pid"], live["writer_pid"],
                              self.report["parent_pid"]}), 3)
        self.assertEqual(data["gate"]["count_v2"], 0)
        self.assertTrue(data["gate"]["admitted"])
        self.assertEqual(trace[-1]["result"]["error"], "job 2: legacy cannot read schema 2")
        self.assertEqual(data["final"]["policy"], {"epoch": 7, "cap": 1})

    def test_unsafe_retains_prepared_bytes_and_precision(self):
        data = self.scenario("UNSAFE_GATE")
        rows = data["final"]["rows"]
        self.assertEqual([row["id"] for row in rows], [1, 2])
        self.assertEqual(rows[1]["hex"], data["trace"][0]["payload_hex"])
        self.assertEqual(lab.decode(bytes.fromhex(rows[1]["hex"]), "bridge"), 1500)
        self.assertEqual(rows[0], data["initial"]["rows"][0])

    def test_fence_rejects_stale_writer_then_accepts_fresh_bridge(self):
        data = self.scenario("FENCED_GATE")
        trace = data["trace"]
        rejection = next(row["result"] for row in trace if row["event"] == "writer_result")
        self.assertEqual(rejection["event"], "rejected")
        self.assertEqual(rejection["error"], "stale epoch: ticket=7 current=8")
        bridge = next(row for row in trace if row["event"] == "bridge_writer_parked")
        self.assertEqual(bridge["ticket"], 8)
        self.assertNotEqual(bridge["pid"], trace[0]["pid"])
        self.assertEqual(trace[-1]["result"]["delays_ms"], [2000, 3000])
        self.assertEqual(data["final"]["policy"], {"epoch": 8, "cap": 1})
        self.assertEqual([json.loads(bytes.fromhex(row["hex"]))["schema"]
                          for row in data["final"]["rows"]], [1, 1])

    def test_residual_refusal_commits_restriction_and_preserves_v2(self):
        data = self.scenario("RESIDUAL_V2")
        self.assertFalse(data["gate"]["admitted"])
        self.assertEqual(data["gate"]["count_v2"], 1)
        self.assertEqual(data["final"]["policy"], {"epoch": 8, "cap": 1})
        rows = data["final"]["rows"]
        self.assertEqual(len(rows), 2)
        self.assertEqual(lab.decode(bytes.fromhex(rows[1]["hex"]), "bridge"), 1500)
        self.assertEqual(data["trace"][-1]["delays_ms"], [2000, 1500])

    def test_other_serialization_writer_commit_precedes_gate(self):
        data = self.scenario("WRITER_FIRST")
        self.assertEqual([row["event"] for row in data["trace"]], [
            "writer_parked_before_insert", "writer_committed_before_gate", "gate_committed"])
        self.assertEqual(data["trace"][1]["checked_policy"], {"epoch": 7, "cap": 2})
        self.assertFalse(data["gate"]["admitted"])
        self.assertEqual(data["gate"]["count_v2"], 1)
        self.assertEqual(data["final"]["policy"], {"epoch": 8, "cap": 1})

    def test_qa_artifacts_and_process_cleanup(self):
        saved = json.loads((self.root / "qa.json").read_text(encoding="utf-8"))
        self.assertEqual(saved["script_sha256"], self.report["script_sha256"])
        self.assertTrue(saved["original_scripts_unchanged"])
        self.assertEqual(len(saved["scenarios"]), 4)
        self.assertIn("PASS 4 scenarios", (self.root / "stdout.txt").read_text(encoding="utf-8"))
        for data in saved["scenarios"].values():
            self.assertTrue(all(code == 0 for code in data["children_exitcodes"]))
            self.assertEqual(data["final"]["integrity"], [["ok"]])


class NegativeCases(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.root = Path(tempfile.mkdtemp(prefix="race-negative-", dir=race.HERE))

    def database(self, name):
        path = self.root / (name + ".db")
        race.initialize(path)
        return path

    def test_current_epoch_does_not_override_write_cap(self):
        path = self.database("schema-cap")
        race.gate(path)
        before = race.inspect(path)
        with self.assertRaisesRegex(lab.ContractError, "schema 2 exceeds write cap 1"):
            race.fenced_insert(path, 8, lab.encode(1500, "native_v2"))
        self.assertEqual(race.inspect(path), before)

    def test_unknown_corrupt_duplicate_and_boolean_schema_deny_admission(self):
        payloads = ["broken-json", '{"schema":3,"delay":2}',
                    '{"schema":1,"schema":1,"delay":2}',
                    '{"schema":true,"delay":2}', '{"schema":1,"delay":"2"}']
        for index, payload in enumerate(payloads):
            with self.subTest(payload=payload):
                path = self.database(f"invalid-{index}")
                with closing(lab.connect(path, "rw")) as db, db:
                    db.execute("INSERT INTO jobs(payload) VALUES (?)", (payload,))
                before = race.inspect(path)["rows"]
                decision = race.gate(path)
                self.assertEqual(decision["count_v2"], 0)
                self.assertFalse(decision["admitted"])
                self.assertTrue(decision["incompatible"])
                self.assertEqual(race.inspect(path)["policy"], {"epoch": 8, "cap": 1})
                self.assertEqual(race.inspect(path)["rows"], before)

    def test_aba_reenable_v2_does_not_restore_old_ticket(self):
        path = self.database("aba")
        stale_payload = lab.encode(1500, "native_v2")
        race.gate(path)  # epoch 7/cap 2 -> epoch 8/cap 1
        with closing(lab.connect(path, "rw")) as db, db:
            db.execute("BEGIN IMMEDIATE")
            db.execute("UPDATE policy SET epoch=epoch+1, cap=2 WHERE id=1")
        # A new connection observes durable epoch 9, not the old epoch 7.
        before = race.inspect(path)
        self.assertEqual(before["policy"], {"epoch": 9, "cap": 2})
        with self.assertRaisesRegex(lab.ContractError, "stale epoch: ticket=7 current=9"):
            race.fenced_insert(path, 7, stale_payload)
        self.assertEqual(race.inspect(path), before)
        self.assertEqual(race.fenced_insert(path, 9, stale_payload)["schema"], 2)

    def test_direct_insert_bypasses_application_fence_and_breaks_legacy(self):
        path = self.database("bypass")
        decision = race.gate(path)
        self.assertTrue(decision["admitted"])
        # Expected counterexample: policy does not enforce itself on raw SQL.
        with closing(lab.connect(path, "rw")) as db, db:
            db.execute("INSERT INTO jobs(payload) VALUES (?)", (lab.encode(1500, "native_v2"),))
        self.assertEqual(race.inspect(path)["policy"], {"epoch": 8, "cap": 1})
        with self.assertRaisesRegex(lab.ContractError, "job 2: legacy cannot read schema 2"):
            lab.worker(path, "legacy", "read")
        self.assertEqual(lab.worker(path, "bridge", "read")["delays_ms"], [2000, 1500])


if __name__ == "__main__":
    unittest.main(verbosity=2)