#!/usr/bin/env python3 """Capture harness for the DDIA Ch. 8 compare-and-set (optimistic concurrency) exercise. Reproduces lock-free optimistic concurrency control on a REAL PostgreSQL 16 server, using the book's compare-and-set pattern: a `version` column guards the write. Two sessions both read `version = 5`, then each run UPDATE docs SET content = ..., version = 6 WHERE id = 1 AND version = 5 Only ONE of them reports `UPDATE 1`; the loser reports `UPDATE 0` (its WHERE no longer matches, because the winner already bumped version to 6) and must retry. A lost update is prevented without any lock. Sub-surprise: even under REPEATABLE READ, an UPDATE's WHERE clause is evaluated against the LATEST COMMITTED row version, NOT the transaction's snapshot. So in one RR transaction a plain SELECT still shows version = 5 (the snapshot), yet the UPDATE ... WHERE version = 5 affects 0 rows (the latest committed version is 6). That EXCEPTION to snapshot visibility is exactly what makes lock-free CAS correct. This is the CAPTURE / VERIFY path, not the reader path. Readers of the exercise type two `psql` sessions by hand (Session A / Session B); this script drives the same two sessions programmatically so the printed transcript is real, verbatim output that the write-up quotes. It is deterministic and idempotent: it drops and recreates its own database `ex_ch08_cas` on the shared, already-running server, seeds it, runs both scenarios, and closes every connection before dropping. Run: uv run --python 3.12 --with "psycopg[binary]" python3 code/compare_and_set.py Substrate (already running; this script does NOT manage containers): PostgreSQL 16 container `ddia-ch08-pg` on localhost:5432, user=postgres, trust auth, maintenance db `study`. """ import psycopg ADMIN_DSN = "host=localhost port=5432 user=postgres dbname=study" DBNAME = "ex_ch08_cas" EX_DSN = f"host=localhost port=5432 user=postgres dbname={DBNAME}" def recreate_database() -> None: """Drop and recreate our own database via an autocommit admin connection.""" with psycopg.connect(ADMIN_DSN, autocommit=True) as admin: admin.execute(f"DROP DATABASE IF EXISTS {DBNAME} WITH (FORCE)") admin.execute(f"CREATE DATABASE {DBNAME}") def seed() -> None: """Create the docs table and seed one row: id=1, content='draft', version=5.""" with psycopg.connect(EX_DSN, autocommit=True) as conn: conn.execute("DROP TABLE IF EXISTS docs") conn.execute( "CREATE TABLE docs (id int PRIMARY KEY, content text, version int)" ) conn.execute("INSERT INTO docs VALUES (1, 'draft', 5)") def reset_doc() -> None: """Return the doc to its seed state between scenarios.""" with psycopg.connect(EX_DSN, autocommit=True) as conn: conn.execute("UPDATE docs SET content = 'draft', version = 5 WHERE id = 1") def read_version(cur) -> int: cur.execute("SELECT version FROM docs WHERE id = 1") return cur.fetchone()[0] def scenario_two_writers_race() -> None: """The main surprise: two writers do compare-and-set; the loser gets UPDATE 0. Both open their own transaction and read version = 5. Session A runs the guarded UPDATE (WHERE version = 5) and commits -> UPDATE 1, version now 6. Session B runs the identical guarded UPDATE -> UPDATE 0: its WHERE no longer matches, so a lost update is prevented without any lock. B must retry. """ print("=== Scenario 1: two writers race, guarded by version (compare-and-set) ===") sa = psycopg.connect(EX_DSN, autocommit=False) sb = psycopg.connect(EX_DSN, autocommit=False) try: ca, cb = sa.cursor(), sb.cursor() # Both sessions read the same starting version. va = read_version(ca) print(f"Session A SELECT version FROM docs WHERE id=1 -> {va}") vb = read_version(cb) print(f"Session B SELECT version FROM docs WHERE id=1 -> {vb}") # Session A: guarded write with the version it read, then commit. ca.execute( "UPDATE docs SET content = 'A wins', version = 6 " "WHERE id = 1 AND version = %s", (va,), ) print(f"Session A UPDATE ... WHERE id=1 AND version={va} -> " f"UPDATE {ca.rowcount}") sa.commit() print("Session A COMMIT") # Session B: identical guarded write with the (now stale) version it read. cb.execute( "UPDATE docs SET content = 'B wins', version = 6 " "WHERE id = 1 AND version = %s", (vb,), ) print(f"Session B UPDATE ... WHERE id=1 AND version={vb} -> " f"UPDATE {cb.rowcount}") sb.commit() print("Session B COMMIT") # Show who actually won. with psycopg.connect(EX_DSN, autocommit=True) as obs: row = obs.execute( "SELECT content, version FROM docs WHERE id = 1" ).fetchone() print(f"final row: content={row[0]!r}, version={row[1]}") finally: sa.close() sb.close() print() def scenario_snapshot_vs_where() -> None: """The sub-surprise: under REPEATABLE READ, the snapshot and the write path disagree -- and Postgres refuses to let the write proceed silently. Session A opens a REPEATABLE READ transaction and takes its snapshot with a first read (version = 5). Session B commits a bump to version = 6. Session A, still in the SAME transaction: * SELECT version -> 5 (its frozen snapshot -- repeatable read holds) * UPDATE ... WHERE version = 5 -> the row it wants to touch was updated by a concurrent transaction that committed. Under REPEATABLE READ, Postgres cannot silently re-evaluate the WHERE against the new row (that would break snapshot isolation), so instead of UPDATE 0 it ABORTS the whole transaction with SQLSTATE 40001 (serialization_failure). You retry. """ print("=== Scenario 2: REPEATABLE READ -- snapshot (5) vs a concurrent commit (6) ===") sa = psycopg.connect(EX_DSN, autocommit=False) try: sa.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ") ca = sa.cursor() # First read fixes Session A's snapshot at version = 5. v_snapshot = read_version(ca) print(f"Session A (REPEATABLE READ) SELECT version -> {v_snapshot} " f"(snapshot taken)") # Session B bumps version 5 -> 6 and commits, in its own transaction. with psycopg.connect(EX_DSN, autocommit=False) as sb: sb.execute( "UPDATE docs SET content = 'B bumped', version = 6 " "WHERE id = 1 AND version = 5" ) sb.commit() print("Session B committed: version 5 -> 6") # Session A, SAME transaction: SELECT still shows the snapshot... v_still = read_version(ca) print(f"Session A SELECT version (same txn) -> {v_still} " f"(still the snapshot -- repeatable read)") # ...but the guarded UPDATE hits a row a concurrent txn already committed. try: ca.execute( "UPDATE docs SET content = 'A wins', version = 6 " "WHERE id = 1 AND version = 5" ) print(f"Session A UPDATE ... WHERE version=5 -> UPDATE {ca.rowcount}") sa.rollback() except psycopg.errors.SerializationFailure as e: print(f"Session A UPDATE ... WHERE version=5 -> ERROR " f"[SQLSTATE {e.sqlstate}]") print(f" {str(e).strip()}") sa.rollback() print("Session A ROLLBACK (transaction aborted; must retry)") finally: sa.close() print() def scenario_read_committed_where() -> None: """Contrast: at READ COMMITTED, the same interleaving does NOT abort -- the UPDATE's WHERE is re-evaluated against the latest committed row and matches 0 rows (UPDATE 0), exactly like the loser in Scenario 1. """ print("=== Scenario 3: READ COMMITTED -- same interleaving gives UPDATE 0 ===") sa = psycopg.connect(EX_DSN, autocommit=False) try: # READ COMMITTED is the default; set explicitly for clarity. sa.execute("SET TRANSACTION ISOLATION LEVEL READ COMMITTED") ca = sa.cursor() v = read_version(ca) print(f"Session A (READ COMMITTED) SELECT version -> {v}") with psycopg.connect(EX_DSN, autocommit=False) as sb: sb.execute( "UPDATE docs SET content = 'B bumped', version = 6 " "WHERE id = 1 AND version = 5" ) sb.commit() print("Session B committed: version 5 -> 6") v_after = read_version(ca) print(f"Session A SELECT version (new statement snapshot) -> {v_after} " f"(READ COMMITTED sees the latest committed)") ca.execute( "UPDATE docs SET content = 'A wins', version = 6 " "WHERE id = 1 AND version = 5" ) print(f"Session A UPDATE ... WHERE version=5 -> UPDATE {ca.rowcount} " f"(latest committed is 6, guard fails -- no lock, no lost update)") sa.rollback() print("Session A ROLLBACK") finally: sa.close() print() def main() -> None: recreate_database() seed() print("One doc: id=1, content='draft', version=5. " "A version column guards every write.\n") reset_doc() scenario_two_writers_race() reset_doc() scenario_snapshot_vs_where() reset_doc() scenario_read_committed_where() print("=== summary ===") print("Scenario 1: both writers read version=5; the winner gets UPDATE 1, " "the loser UPDATE 0 -- a lost update prevented lock-free.") print("Scenario 2: under REPEATABLE READ, SELECT still shows the snapshot (5) " "after a concurrent commit, and the guarded UPDATE aborts with " "SQLSTATE 40001 (serialization_failure) rather than silently no-op.") print("Scenario 3: under READ COMMITTED, the same interleaving gives UPDATE 0 " "-- the WHERE is re-checked against the latest committed row (now 6).") if __name__ == "__main__": main()