附录：随书实验完整代码

"""Real MySQL 8.4 experiments; no in-memory substitute database."""

"""A loopback, plaintext MySQL packet relay that drops exactly a COMMIT reply.

This is a fault injector, not a SQL proxy or a production authentication layer.
The server's OK packet is observed before the client socket is closed. No packet
contents or credentials are logged; TLS connections are outside this lab.
"""
import socket
import socketserver
import threading


def packet(sock):
    def exact(size):
        result = bytearray()
        while len(result) < size:
            block = sock.recv(size - len(result))
            if not block:
                raise EOFError
            result.extend(block)
        return bytes(result)
    header = exact(4)
    length = int.from_bytes(header[:3], "little")
    if length > 1024 * 1024:
        raise ValueError("lab_packet_limit")
    return header, exact(length)


class CommitReplyDropper:
    def __init__(self, upstream_host, upstream_port):
        if upstream_host not in {"127.0.0.1", "localhost", "::1"}:
            raise ValueError("loopback_only")
        self.commit_ok_seen = threading.Event()
        owner = self

        class Server(socketserver.ThreadingTCPServer):
            allow_reuse_address = True
            daemon_threads = True

        class Handler(socketserver.BaseRequestHandler):
            def handle(self):
                upstream = socket.create_connection((upstream_host, upstream_port), timeout=10)
                downstream = self.request
                downstream.settimeout(10)
                drop_next_commit = threading.Event()

                def close_both():
                    for stream in (upstream, downstream):
                        try:
                            stream.shutdown(socket.SHUT_RDWR)
                        except OSError:
                            pass

                def server_to_client():
                    try:
                        while True:
                            header, payload = packet(upstream)
                            if drop_next_commit.is_set() and payload[:1] == b"\x00":
                                owner.commit_ok_seen.set()
                                close_both()
                                return
                            downstream.sendall(header + payload)
                    except (EOFError, OSError, ValueError):
                        close_both()

                thread = threading.Thread(target=server_to_client, daemon=True)
                thread.start()
                try:
                    while True:
                        header, payload = packet(downstream)
                        if payload[:1] == b"\x03" and payload[1:].strip().upper() == b"COMMIT":
                            drop_next_commit.set()
                        upstream.sendall(header + payload)
                except (EOFError, OSError, ValueError):
                    pass
                finally:
                    close_both()
                    upstream.close()
                    thread.join(2)

        self.server = Server(("127.0.0.1", 0), Handler)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(target=lambda: self.server.serve_forever(poll_interval=.02), daemon=True)

    def __enter__(self):
        self.thread.start()
        return self

    def __exit__(self, *_):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(2)

"""Connections and uniquely owned disposable schemas for the book experiments."""
import os
import re
import uuid
import pymysql


def settings():
    host = os.environ.get("MYSQL_HOST", "127.0.0.1")
    if host not in {"127.0.0.1", "localhost", "::1"}:
        raise ValueError("Use the isolated local lab or a local port-forward, not an arbitrary server.")
    return dict(host=host, port=int(os.environ.get("MYSQL_PORT", "23306")),
                user=os.environ.get("MYSQL_USER", "root"),
                password=os.environ.get("MYSQL_PASSWORD", "book-lab-only"),
                charset="utf8mb4", connect_timeout=10, read_timeout=12, write_timeout=5)


def connect(schema=None, autocommit=False, isolation="REPEATABLE READ", port=None):
    config = settings()
    if port is not None:
        config["port"] = port
        # Only the loopback packet fault injector needs plaintext. PyMySQL
        # 1.2 negotiates TLS by default when the real server supports it.
        config["ssl_disabled"] = True
    connection = pymysql.connect(**config, database=schema, autocommit=autocommit)
    if isolation not in {"REPEATABLE READ", "READ COMMITTED"}:
        connection.close()
        raise ValueError("unsupported lab isolation")
    with connection.cursor() as cursor:
        cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL " + isolation)
        cursor.execute("SET SESSION innodb_lock_wait_timeout=3")
    return connection


def scalar(connection, sql, args=None):
    with connection.cursor() as cursor:
        cursor.execute(sql, args)
        return cursor.fetchone()[0]


class OwnedSchema:
    """Never drops a database supplied by a caller or environment variable."""
    def __init__(self):
        self.name = "book_lab_" + uuid.uuid4().hex[:16]
        self.created = False
        with connect(autocommit=True) as connection:
            with connection.cursor() as cursor:
                cursor.execute("CREATE DATABASE `" + self.name + "` CHARACTER SET utf8mb4")
        self.created = True

    def connection(self, **kwargs):
        return connect(self.name, **kwargs)

    def close(self):
        if self.created:
            assert re.fullmatch(r"book_lab_[0-9a-f]{16}", self.name)
            with connect(autocommit=True) as connection:
                with connection.cursor() as cursor:
                    cursor.execute("DROP DATABASE `" + self.name + "`")
            self.created = False

"""Run real scenarios once and save their observations, including failures."""
import argparse
from datetime import datetime, timezone
from decimal import Decimal
import importlib.metadata
import json
from pathlib import Path
import platform
import unittest

from .scenarios import InnoDBCases, EVIDENCE


def exact_json(value):
    if isinstance(value, Decimal):
        return int(value) if value == value.to_integral_value() else {"decimal": str(value)}
    raise TypeError("Unsupported evidence type: " + type(value).__name__)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--evidence", default="lab/evidence/mysql.json")
    arguments = parser.parse_args()
    EVIDENCE.clear()
    suite = unittest.defaultTestLoader.loadTestsFromTestCase(InnoDBCases)
    result = unittest.TextTestRunner(verbosity=2).run(suite)
    report = dict(tested_at=datetime.now(timezone.utc).isoformat(), python=platform.python_version(),
                  driver_version=importlib.metadata.version("PyMySQL"), tests_run=result.testsRun,
                  passed=result.wasSuccessful(), observations=EVIDENCE,
                  failures=[dict(test=str(case), traceback=message) for case,message in result.failures+result.errors])
    target = Path(arguments.evidence)
    target.parent.mkdir(parents=True,exist_ok=True)
    target.write_text(json.dumps(report,ensure_ascii=False,indent=2,default=exact_json)+"\n")
    raise SystemExit(0 if result.wasSuccessful() else 1)


if __name__ == "__main__":
    main()

["mysql@sha256:0f51e63e537a1f176423fa03ef559aabe6fa4372a53160a44bc7e75da7d15364"]

{
  "tested_at": "2026-10-21T09:53:27.447278+00:00",
  "python": "3.12.15",
  "driver_version": "1.2.3",
  "tests_run": 15,
  "passed": true,
  "observations": [
    {
      "scenario": "test_01_server_contract",
      "version": "8.4.11",
      "isolation": "REPEATABLE-READ",
      "innodb_rollback_on_timeout": 0,
      "deadlock_detection": 1
    },
    {
      "scenario": "test_02_rollback_and_savepoint_are_transactional",
      "committed_after_savepoint": [
        80,
        100
      ],
      "balance_after_full_rollback": 80
    },
    {
      "scenario": "test_03_rr_snapshot_starts_at_first_consistent_read",
      "after_begin_before_first_read": 2,
      "same_snapshot": 2,
      "new_transaction": 3
    },
    {
      "scenario": "test_04_rc_uses_fresh_committed_snapshot",
      "first": 1,
      "second": 2
    },
    {
      "scenario": "test_05_rr_locking_read_can_see_a_newer_row",
      "consistent_before": 1,
      "locking_read": 2,
      "consistent_after": 1
    },
    {
      "scenario": "test_06_read_then_absolute_write_violates_inventory_invariant",
      "initial_stock": 1,
      "remaining": 0,
      "sold": 2,
      "invariant_satisfied": false
    },
    {
      "scenario": "test_07_conditional_update_preserves_inventory_invariant",
      "initial_stock": 1,
      "remaining": 0,
      "sold": 1,
      "invariant_satisfied": true
    },
    {
      "scenario": "test_08_unique_operation_key_prevents_duplicate_purchase",
      "outcomes": [
        "committed",
        "duplicate"
      ],
      "orders": 1,
      "remaining": 0
    },
    {
      "scenario": "test_09_deadlock_rolls_back_one_entire_transaction",
      "outcomes": [
        "committed",
        "deadlock_1213"
      ],
      "balances": [
        101,
        101
      ],
      "victim_final_action": "commit_without_explicit_rollback"
    },
    {
      "scenario": "test_10_lock_timeout_does_not_rollback_prior_statement_by_default",
      "error": 1205,
      "balances_after_intentional_commit": [
        100,
        105
      ],
      "lesson": "rollback the application transaction before retrying"
    },
    {
      "scenario": "test_11_index_changes_plan_without_claiming_a_speed_benchmark",
      "before": {
        "table_name": "people",
        "access_type": "ALL",
        "rows_examined_per_scan": 2000,
        "rows_produced_per_join": 200,
        "filtered": "10.00",
        "cost_info": {
          "read_cost": "180.25",
          "eval_cost": "20.00",
          "prefix_cost": "200.25",
          "data_read_per_join": "76K"
        },
        "used_columns": [
          "id",
          "email"
        ],
        "attached_condition": "(`book_lab_26d6d2152cbb4fb0`.`people`.`email` = 'user1000@example.invalid')"
      },
      "after": {
        "table_name": "people",
        "access_type": "const",
        "possible_keys": [
          "email_idx"
        ],
        "key": "email_idx",
        "used_key_parts": [
          "email"
        ],
        "key_length": "386",
        "ref": [
          "const"
        ],
        "rows_examined_per_scan": 1,
        "rows_produced_per_join": 1,
        "filtered": "100.00",
        "using_index": true,
        "cost_info": {
          "read_cost": "0.00",
          "eval_cost": "0.10",
          "prefix_cost": "0.00",
          "data_read_per_join": "392"
        },
        "used_columns": [
          "id",
          "email"
        ]
      },
      "note": "execution plan, not a latency benchmark"
    },
    {
      "scenario": "test_12_order_and_outbox_share_one_local_transaction",
      "committed_counts": [
        1,
        1
      ],
      "scope": "database local atomicity; no broker delivery claim"
    },
    {
      "scenario": "test_13_skip_locked_claims_distinct_rows_and_rollback_releases_claim",
      "first": 1,
      "concurrent_claim": 2,
      "after_rollback": 1
    },
    {
      "scenario": "test_14_commit_reply_lost_can_still_mean_committed",
      "client_error_code": 2013,
      "server_commit_ok_observed": true,
      "independently_found_order": 1,
      "lesson": "query stable business identity before replay"
    },
    {
      "scenario": "test_15_performance_schema_exposes_real_blocking_edge",
      "observed_lock_wait_edges": 1,
      "blocker_rollback_released_waiter": true
    }
  ],
  "failures": []
}

"""Real InnoDB experiments. Every assertion queries independent database state."""
import json
import threading
import time
import unittest
import pymysql

from .commit_proxy import CommitReplyDropper
from .db import OwnedSchema, scalar, settings

EVIDENCE = []


class InnoDBCases(unittest.TestCase):
    def setUp(self):
        self.schema = OwnedSchema()
        self.connections = []
        self.control = self.open(autocommit=True)
        with self.control.cursor() as cursor:
            cursor.execute("CREATE TABLE inventory(sku INT PRIMARY KEY,stock INT NOT NULL) ENGINE=InnoDB")
            cursor.execute("INSERT INTO inventory VALUES(1,1)")
            cursor.execute("CREATE TABLE orders(op_id VARBINARY(64) PRIMARY KEY,qty INT NOT NULL) ENGINE=InnoDB")
            cursor.execute("CREATE TABLE accounts(id INT PRIMARY KEY,balance INT NOT NULL) ENGINE=InnoDB")
            cursor.execute("INSERT INTO accounts VALUES(1,100),(2,100)")
            cursor.execute("CREATE TABLE outbox(id BIGINT AUTO_INCREMENT PRIMARY KEY,payload JSON NOT NULL,sent BOOLEAN NOT NULL DEFAULT FALSE) ENGINE=InnoDB")
        self.c = self.open()

    def tearDown(self):
        for connection in self.connections:
            try:
                connection.rollback()
                connection.close()
            except pymysql.MySQLError:
                pass
        self.schema.close()

    def open(self, **kwargs):
        connection = self.schema.connection(**kwargs)
        self.connections.append(connection)
        return connection

    def record(self, **data):
        EVIDENCE.append(dict(scenario=self._testMethodName, **data))

    def parallel(self, functions):
        errors = []
        def wrapper(function):
            try:
                function()
            except BaseException as error:
                errors.append(error)
        threads = [threading.Thread(target=wrapper, args=(f,), daemon=True) for f in functions]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join(10)
        self.assertFalse(any(t.is_alive() for t in threads), "experiment thread did not stop")
        if errors:
            raise errors[0]

    def test_01_server_contract(self):
        version = scalar(self.c, "SELECT VERSION()")
        isolation = scalar(self.c, "SELECT @@transaction_isolation")
        rollback_timeout = scalar(self.c, "SELECT @@innodb_rollback_on_timeout")
        deadlock = scalar(self.c, "SELECT @@innodb_deadlock_detect")
        self.assertTrue(version.startswith("8.4."), version)
        self.assertEqual(isolation, "REPEATABLE-READ")
        self.assertEqual(rollback_timeout, 0)
        self.assertEqual(deadlock, 1)
        self.record(version=version, isolation=isolation,
                    innodb_rollback_on_timeout=rollback_timeout, deadlock_detection=deadlock)

    def test_02_rollback_and_savepoint_are_transactional(self):
        with self.c.cursor() as cursor:
            cursor.execute("UPDATE accounts SET balance=80 WHERE id=1")
            cursor.execute("SAVEPOINT before_second")
            cursor.execute("UPDATE accounts SET balance=120 WHERE id=2")
            cursor.execute("ROLLBACK TO SAVEPOINT before_second")
        self.c.commit()
        self.assertEqual(scalar(self.control, "SELECT balance FROM accounts WHERE id=1"), 80)
        self.assertEqual(scalar(self.control, "SELECT balance FROM accounts WHERE id=2"), 100)
        self.c.begin()
        with self.c.cursor() as cursor:
            cursor.execute("UPDATE accounts SET balance=1 WHERE id=1")
        self.c.rollback()
        self.assertEqual(scalar(self.control, "SELECT balance FROM accounts WHERE id=1"), 80)
        self.record(committed_after_savepoint=[80, 100], balance_after_full_rollback=80)

    def test_03_rr_snapshot_starts_at_first_consistent_read(self):
        self.c.begin()
        other = self.open()
        with other.cursor() as cursor:
            cursor.execute("UPDATE inventory SET stock=2 WHERE sku=1")
        other.commit()
        first = scalar(self.c, "SELECT stock FROM inventory WHERE sku=1")
        with other.cursor() as cursor:
            cursor.execute("UPDATE inventory SET stock=3 WHERE sku=1")
        other.commit()
        second = scalar(self.c, "SELECT stock FROM inventory WHERE sku=1")
        self.c.commit()
        fresh = scalar(self.c, "SELECT stock FROM inventory WHERE sku=1")
        self.assertEqual((first, second, fresh), (2, 2, 3))
        self.record(after_begin_before_first_read=first, same_snapshot=second, new_transaction=fresh)

    def test_04_rc_uses_fresh_committed_snapshot(self):
        reader = self.open(isolation="READ COMMITTED")
        first = scalar(reader, "SELECT stock FROM inventory WHERE sku=1")
        with self.control.cursor() as cursor:
            cursor.execute("UPDATE inventory SET stock=2 WHERE sku=1")
        second = scalar(reader, "SELECT stock FROM inventory WHERE sku=1")
        self.assertEqual((first, second), (1, 2))
        self.record(first=first, second=second)

    def test_05_rr_locking_read_can_see_a_newer_row(self):
        before = scalar(self.c, "SELECT stock FROM inventory WHERE sku=1")
        with self.control.cursor() as cursor:
            cursor.execute("UPDATE inventory SET stock=2 WHERE sku=1")
        locked = scalar(self.c, "SELECT stock FROM inventory WHERE sku=1 FOR UPDATE")
        consistent = scalar(self.c, "SELECT stock FROM inventory WHERE sku=1")
        self.assertEqual((before, locked, consistent), (1, 2, 1))
        self.c.rollback()
        self.record(consistent_before=before, locking_read=locked, consistent_after=consistent)

    def test_06_read_then_absolute_write_violates_inventory_invariant(self):
        ready = threading.Barrier(2, timeout=5)
        def buyer(key):
            connection = self.open()
            amount = scalar(connection, "SELECT stock FROM inventory WHERE sku=1")
            ready.wait()
            if amount > 0:
                with connection.cursor() as cursor:
                    cursor.execute("UPDATE inventory SET stock=%s WHERE sku=1", (amount-1,))
                    cursor.execute("INSERT INTO orders VALUES(%s,1)", (key,))
                connection.commit()
        self.parallel([lambda: buyer(b"a"), lambda: buyer(b"b")])
        stock = scalar(self.control, "SELECT stock FROM inventory WHERE sku=1")
        sales = scalar(self.control, "SELECT SUM(qty) FROM orders")
        self.assertEqual((stock, sales), (0, 2))
        self.record(initial_stock=1, remaining=stock, sold=sales, invariant_satisfied=False)

    def test_07_conditional_update_preserves_inventory_invariant(self):
        ready = threading.Barrier(2, timeout=5)
        def buyer(key):
            connection = self.open()
            ready.wait()
            with connection.cursor() as cursor:
                changed = cursor.execute("UPDATE inventory SET stock=stock-1 WHERE sku=1 AND stock>0")
                if changed:
                    cursor.execute("INSERT INTO orders VALUES(%s,1)", (key,))
            connection.commit()
        self.parallel([lambda: buyer(b"a"), lambda: buyer(b"b")])
        stock = scalar(self.control, "SELECT stock FROM inventory WHERE sku=1")
        sales = scalar(self.control, "SELECT SUM(qty) FROM orders")
        self.assertEqual((stock, sales), (0, 1))
        self.record(initial_stock=1, remaining=stock, sold=sales, invariant_satisfied=True)

    def test_08_unique_operation_key_prevents_duplicate_purchase(self):
        ready = threading.Barrier(2, timeout=5)
        outcomes = []
        def buyer():
            connection = self.open()
            ready.wait()
            try:
                with connection.cursor() as cursor:
                    cursor.execute("INSERT INTO orders VALUES(%s,1)", (b"same-operation",))
                    changed = cursor.execute("UPDATE inventory SET stock=stock-1 WHERE sku=1 AND stock>0")
                    if changed != 1:
                        raise AssertionError("inventory rejected")
                connection.commit()
                outcomes.append("committed")
            except pymysql.err.IntegrityError as error:
                self.assertEqual(error.args[0], 1062)
                connection.rollback()
                outcomes.append("duplicate")
        self.parallel([buyer, buyer])
        self.assertCountEqual(outcomes, ["committed", "duplicate"])
        self.assertEqual(scalar(self.control, "SELECT COUNT(*) FROM orders"), 1)
        self.assertEqual(scalar(self.control, "SELECT stock FROM inventory WHERE sku=1"), 0)
        self.record(outcomes=sorted(outcomes), orders=1, remaining=0)

    def test_09_deadlock_rolls_back_one_entire_transaction(self):
        ready = threading.Barrier(2, timeout=5)
        outcomes = []
        def actor(first, second):
            connection = self.open()
            try:
                with connection.cursor() as cursor:
                    cursor.execute("UPDATE accounts SET balance=balance+1 WHERE id=%s", (first,))
                    ready.wait()
                    cursor.execute("UPDATE accounts SET balance=balance+1 WHERE id=%s", (second,))
                connection.commit()
                outcomes.append("committed")
            except pymysql.err.OperationalError as error:
                self.assertEqual(error.args[0], 1213)
                # Deliberately COMMIT without an explicit ROLLBACK: if the engine
                # had retained the first update, it would become visible below.
                # This is an isolated probe of automatic rollback, not recovery advice.
                connection.commit()
                outcomes.append("deadlock_1213")
        self.parallel([lambda: actor(1,2), lambda: actor(2,1)])
        balances = [scalar(self.control, "SELECT balance FROM accounts WHERE id=%s", (i,)) for i in (1,2)]
        self.assertCountEqual(outcomes, ["committed", "deadlock_1213"])
        self.assertEqual(balances, [101,101])
        self.record(outcomes=sorted(outcomes), balances=balances,
                    victim_final_action="commit_without_explicit_rollback")

    def test_10_lock_timeout_does_not_rollback_prior_statement_by_default(self):
        holder = self.open()
        with holder.cursor() as cursor:
            cursor.execute("UPDATE accounts SET balance=balance+1 WHERE id=1")
        with self.c.cursor() as cursor:
            cursor.execute("SET SESSION innodb_lock_wait_timeout=1")
            cursor.execute("UPDATE accounts SET balance=balance+5 WHERE id=2")
            with self.assertRaises(pymysql.err.OperationalError) as error:
                cursor.execute("UPDATE accounts SET balance=balance+5 WHERE id=1")
        self.assertEqual(error.exception.args[0], 1205)
        self.c.commit()  # Intentionally shows the partial-commit pitfall in an isolated lab.
        holder.rollback()
        balances = [scalar(self.control, "SELECT balance FROM accounts WHERE id=%s", (i,)) for i in (1,2)]
        self.assertEqual(balances, [100,105])
        self.record(error=1205, balances_after_intentional_commit=balances,
                    lesson="rollback the application transaction before retrying")

    def test_11_index_changes_plan_without_claiming_a_speed_benchmark(self):
        with self.control.cursor() as cursor:
            cursor.execute("CREATE TABLE people(id INT PRIMARY KEY,email VARCHAR(96) NOT NULL) ENGINE=InnoDB")
            cursor.executemany("INSERT INTO people VALUES(%s,%s)", [(i, f"user{i}@example.invalid") for i in range(2000)])
        sql = "EXPLAIN FORMAT=JSON SELECT id FROM people WHERE email='user1000@example.invalid'"
        before = json.loads(scalar(self.control, sql))
        with self.control.cursor() as cursor:
            cursor.execute("ALTER TABLE people ADD UNIQUE KEY email_idx(email)")
        after = json.loads(scalar(self.control, sql))
        def table(node):
            if isinstance(node, dict):
                if node.get("table_name") == "people":
                    return node
                for value in node.values():
                    found = table(value)
                    if found:
                        return found
            elif isinstance(node, list):
                for value in node:
                    found = table(value)
                    if found:
                        return found
        old, new = table(before), table(after)
        self.assertEqual(old["access_type"], "ALL")
        self.assertEqual(new["key"], "email_idx")
        self.assertIn(new["access_type"], {"const", "ref"})
        self.record(before=old, after=new, note="execution plan, not a latency benchmark")

    def test_12_order_and_outbox_share_one_local_transaction(self):
        def create(key, commit):
            with self.c.cursor() as cursor:
                cursor.execute("INSERT INTO orders VALUES(%s,1)", (key,))
                cursor.execute("INSERT INTO outbox(payload) VALUES(%s)", (json.dumps({"operation":key.decode()}),))
            self.c.commit() if commit else self.c.rollback()
        create(b"kept", True)
        create(b"rolled-back", False)
        counts = [scalar(self.control, "SELECT COUNT(*) FROM " + name) for name in ("orders","outbox")]
        self.assertEqual(counts, [1,1])
        self.record(committed_counts=counts, scope="database local atomicity; no broker delivery claim")

    def test_13_skip_locked_claims_distinct_rows_and_rollback_releases_claim(self):
        with self.control.cursor() as cursor:
            cursor.executemany("INSERT INTO outbox(payload) VALUES(%s)", [("{}",),("{}",)])
        first = scalar(self.c, "SELECT id FROM outbox WHERE sent=FALSE ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED")
        other = self.open()
        second = scalar(other, "SELECT id FROM outbox WHERE sent=FALSE ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED")
        self.assertEqual((first, second), (1,2))
        self.c.rollback()
        third = self.open()
        recovered = scalar(third, "SELECT id FROM outbox WHERE sent=FALSE ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED")
        self.assertEqual(recovered, 1)
        other.rollback()
        third.rollback()
        self.record(first=first, concurrent_claim=second, after_rollback=recovered)

    def test_14_commit_reply_lost_can_still_mean_committed(self):
        config = settings()
        with CommitReplyDropper(config["host"], config["port"]) as proxy:
            writer = self.open(port=proxy.port)
            with writer.cursor() as cursor:
                cursor.execute("INSERT INTO orders VALUES(%s,1)", (b"commit-unknown",))
            with self.assertRaises(pymysql.err.OperationalError) as error:
                writer.commit()
            self.assertTrue(proxy.commit_ok_seen.wait(2), "fault must happen after server COMMIT OK")
        actual = scalar(self.control, "SELECT COUNT(*) FROM orders WHERE op_id=%s", (b"commit-unknown",))
        self.assertEqual(actual, 1)
        self.record(client_error_code=error.exception.args[0], server_commit_ok_observed=True,
                    independently_found_order=actual, lesson="query stable business identity before replay")

    def test_15_performance_schema_exposes_real_blocking_edge(self):
        holder = self.open()
        with holder.cursor() as cursor:
            cursor.execute("UPDATE accounts SET balance=balance+1 WHERE id=1")
        begun, stopped = threading.Event(), threading.Event()
        failures = []
        def waiting_writer():
            connection = self.open()
            try:
                begun.set()
                with connection.cursor() as cursor:
                    cursor.execute("UPDATE accounts SET balance=balance+1 WHERE id=1")
                connection.commit()
            except BaseException as error:
                failures.append(error)
            finally:
                stopped.set()
        thread = threading.Thread(target=waiting_writer, daemon=True)
        thread.start()
        observed = 0
        try:
            self.assertTrue(begun.wait(2))
            deadline = time.monotonic()+2
            while time.monotonic()<deadline and not observed:
                observed = scalar(self.control, """
                    SELECT COUNT(*) FROM performance_schema.data_lock_waits w
                    JOIN performance_schema.data_locks r
                      ON r.ENGINE=w.ENGINE AND r.ENGINE_LOCK_ID=w.REQUESTING_ENGINE_LOCK_ID
                    WHERE r.OBJECT_SCHEMA=%s AND r.OBJECT_NAME='accounts'
                    """, (self.schema.name,))
                if not observed:
                    time.sleep(.01)
            self.assertGreaterEqual(observed, 1, "must observe an actual database lock wait")
        finally:
            holder.rollback()
            thread.join(8)
        self.assertTrue(stopped.is_set())
        if failures:
            raise failures[0]
        self.assertEqual(scalar(self.control, "SELECT balance FROM accounts WHERE id=1"), 101)
        self.record(observed_lock_wait_edges=observed, blocker_rollback_released_waiter=True)

from lab.scenarios import InnoDBCases