"""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)
