mvcc-isolation-tests
Per the PostgreSQL transaction isolation docs, the four standard
SQL isolation levels permit different anomalies. Production code
must run at the right level - and tests verify both that the
chosen level prevents the anomalies the business cares about AND
that the code copes when the level allows them.
When to use
- Banking, inventory, booking - anywhere two transactions can
conflict on the same data.
- Choosing or auditing default isolation level for a new service.
- Migration: changing isolation level (e.g., Read Committed →
Repeatable Read) - tests verify nothing relies on old behavior.
Step 1 - Anomaly catalog
Per the PostgreSQL transaction isolation docs:
| Anomaly | What |
|---|
| Dirty read | T2 reads uncommitted T1 changes |
| Non-repeatable read | T1 reads X = 5; T2 commits X = 6; T1 re-reads X = 6 |
| Phantom read | T1 query returns 3 rows; T2 inserts row matching predicate; T1 re-runs query, gets 4 rows |
| Serialization anomaly | Concurrent execution produces a result no serial order would |
| Write skew | T1 reads {x, y} = {0, 0}; T2 reads same; both decide based on read-set; both commit; final state inconsistent |
Step 2 - PostgreSQL isolation matrix
Per the PostgreSQL transaction isolation docs:
| Level | Dirty | Non-repeatable | Phantom | Serialization |
|---|
| Read Uncommitted (= Read Committed in PG) | NO | YES | YES | YES |
| Read Committed | NO | YES | YES | YES |
| Repeatable Read | NO | NO | NO (PG strict) | YES |
| Serializable | NO | NO | NO | NO |
PostgreSQL's Repeatable Read is stricter than SQL standard - it
prevents phantoms via snapshot isolation. Other databases differ.
Step 3 - Test setup: two-connection harness
import psycopg
import threading
import time # Steps 4-6 use time.sleep for the interleaving window
def two_connection_test(workload_a, workload_b, isolation="read committed"):
conn_a = psycopg.connect("dbname=test", autocommit=False)
conn_b = psycopg.connect("dbname=test", autocommit=False)
# Set isolation level for both
for c in (conn_a, conn_b):
with c.cursor() as cur:
cur.execute(f"SET TRANSACTION ISOLATION LEVEL {isolation}")
barrier = threading.Barrier(2)
results = {}
def run(conn, work, key):
barrier.wait()
results[key] = work(conn)
t_a = threading.Thread(target=run, args=(conn_a, workload_a, "a"))
t_b = threading.Thread(target=run, args=(conn_b, workload_b, "b"))
t_a.start(); t_b.start()
t_a.join(); t_b.join()
return results
The Steps 4-6 workloads use time.sleep(0.5) to open the interleaving
window, which is timing-sensitive and can flake on slow CI. For a
deterministic harness, replace the sleep with a second barrier (or a
threading.Event) so the writer commits exactly between the reader's two
reads, no wall-clock guess required:
# Deterministic variant: pass an extra Barrier(2) into both workloads.
# reader: read #1 -> gate.wait() -> read #2
# writer: gate.wait() -> UPDATE + commit (writer races read #2)
# Use a pair of barriers (read1-done, write-done) if you need to pin the
# writer's commit strictly after read #1 and strictly before read #2.
gate = threading.Barrier(2)
Step 4 - Test for non-repeatable read at Read Committed
def test_non_repeatable_read_under_read_committed():
setup_balance(account="acc1", balance=100)
def reader(conn):
with conn.cursor() as cur:
cur.execute("SELECT balance FROM accounts WHERE id = 'acc1'")
first = cur.fetchone()[0]
time.sleep(0.5) # let writer commit
cur.execute("SELECT balance FROM accounts WHERE id = 'acc1'")
second = cur.fetchone()[0]
conn.commit()
return (first, second)
def writer(conn):
with conn.cursor() as cur:
cur.execute("UPDATE accounts SET balance = 200 WHERE id = 'acc1'")
conn.commit()
return None
results = two_connection_test(reader, writer, isolation="read committed")
first, second = results["a"]
assert first == 100
assert second == 200, "Read Committed permits non-repeatable read"
Step 5 - Test phantom read prevented at Repeatable Read (PG)
def test_phantom_read_prevented_under_repeatable_read():
setup_orders(initial_count=3)
def reader(conn):
with conn.cursor() as cur:
cur.execute("SELECT COUNT(*) FROM orders WHERE status = 'pending'")
first_count = cur.fetchone()[0]
time.sleep(0.5)
cur.execute("SELECT COUNT(*) FROM orders WHERE status = 'pending'")
second_count = cur.fetchone()[0]
conn.commit()
return (first_count, second_count)
def writer(conn):
with conn.cursor() as cur:
cur.execute("INSERT INTO orders (status) VALUES ('pending')")
conn.commit()
return None
results = two_connection_test(reader, writer, isolation="repeatable read")
first, second = results["a"]
assert first == second == 3, "PG Repeatable Read should prevent phantom"
Step 6 - Test write-skew at Repeatable Read (still possible!)
Per the PostgreSQL transaction isolation docs, write skew is a
serialization anomaly that Repeatable Read does NOT prevent:
def test_write_skew_under_repeatable_read():
"""Two doctors on call; rule: at least one must remain on call.
T1 reads {alice: on, bob: on}; decides safe to take alice off.
T2 reads same; decides safe to take bob off.
Both commit. Now nobody on call → anomaly."""
setup_doctors([("alice", "on_call"), ("bob", "on_call")])
def take_alice_off(conn):
with conn.cursor() as cur:
cur.execute("SELECT COUNT(*) FROM doctors WHERE status = 'on_call'")
on_call_count = cur.fetchone()[0]
if on_call_count >= 2:
cur.execute("UPDATE doctors SET status = 'off' WHERE name = 'alice'")
conn.commit()
def take_bob_off(conn):
with conn.cursor() as cur:
cur.execute("SELECT COUNT(*) FROM doctors WHERE status = 'on_call'")
on_call_count = cur.fetchone()[0]
if on_call_count >= 2:
cur.execute("UPDATE doctors SET status = 'off' WHERE name = 'bob'")
conn.commit()
two_connection_test(take_alice_off, take_bob_off, isolation="repeatable read")
on_call_after = count_on_call()
# Under Repeatable Read, write skew is possible: both txns read 2
# on-call, both flip one off, and the final state is 0 on call (the
# anomaly). A correctly-serialized run would leave 1. Either value
# proves RR does NOT serialize this pair, so both are accepted here.
# (Run the same workload at Serializable - Step 7 - to see one txn abort.)
assert on_call_after in (0, 1)
Step 7 - Test Serializable rolls back conflicting transactions
def test_serializable_aborts_one_of_skew_pair():
setup_doctors([("alice", "on_call"), ("bob", "on_call")])
aborted = [0]
def take_off(name, conn):
try:
with conn.cursor() as cur:
cur.execute("SELECT COUNT(*) FROM doctors WHERE status = 'on_call'")
if cur.fetchone()[0] >= 2:
cur.execute("UPDATE doctors SET status = 'off' WHERE name = %s", (name,))
conn.commit()
except psycopg.errors.SerializationFailure:
conn.rollback()
aborted[0] += 1
two_connection_test(
lambda c: take_off("alice", c),
lambda c: take_off("bob", c),
isolation="serializable",
)
assert aborted[0] >= 1, "Serializable must abort at least one"
assert count_on_call() >= 1, "At least one doctor still on call"
Per the PostgreSQL transaction isolation docs: Serializable raises
could not serialize access due to read/write dependencies among transactions - application must catch + retry.
Step 8 - Per-database differences
| DB | Default | RR prevents phantoms? | Serializable cost |
|---|
| PostgreSQL | Read Committed | Yes (snapshot isolation) | Predicate locks; can be expensive |
| MySQL InnoDB | Repeatable Read | Yes (gap locks) | Default RR is "good enough" for most |
| SQL Server | Read Committed | No (default), Yes with READ_COMMITTED_SNAPSHOT | Snapshot isolation available |
| Oracle | Read Committed | Yes via Read Consistency | Serializable via lock |
| DynamoDB | Eventually consistent reads | Strongly consistent reads opt-in | Transactions: ACID with TransactWriteItems |
| MongoDB | Snapshot read concern | Multi-doc txn (4.0+) ACID | Limit on doc size (16MB / txn) |
Test the actual DB you ship with - defaults differ.
Step 9 - Application-level retry pattern
def with_serializable_retry(fn, max_retries=3):
for attempt in range(max_retries):
try:
with conn.cursor() as cur:
cur.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
result = fn()
conn.commit()
return result
except psycopg.errors.SerializationFailure:
conn.rollback()
if attempt == max_retries - 1:
raise
time.sleep(0.01 * (2 ** attempt))
Test the retry logic itself:
def test_serializable_retry_completes():
# Simulate concurrent conflict; verify retry succeeds eventually
...
Anti-patterns
| Anti-pattern | Why it fails | Fix |
|---|
| Run at default isolation, hope for the best | Anomaly hits in prod | Choose level deliberately + test |
| Test only happy-path single-transaction | Concurrency bugs need 2-conn harness | Step 3 harness |
| Use Serializable everywhere "to be safe" | Performance cliff; retries swamp | Test perf at chosen level |
| Skip retry logic for Serializable | Production transaction errors leak | Step 9 retry pattern |
| Test PG and assume MySQL behaves same | RR semantics differ (gap locks vs snapshot) | Per-DB tests (Step 8) |
Limitations
- Test fixture flakiness: timing-sensitive; may need barrier
adjustment per machine speed.
- Eventually-consistent NoSQL stores need different tests
(probabilistic + windowed).
- ORM abstractions can hide explicit SET ISOLATION LEVEL - verify
the actual SQL emitted.
References
- PostgreSQL transaction isolation docs - anomaly definitions,
per-level matrix, retry patterns
- MySQL InnoDB isolation - dev.mysql.com/doc/refman/8.0/en/innodb-transaction-isolation-levels.html
- A Critique of ANSI SQL Isolation Levels (Berenson 1995) - seminal
paper on isolation levels
race-condition-test-author,
deadlock-detection-harness -
in-process concurrency siblings