Map Postgres errors onto a DBAPI2 exception hierarchy

Previously every `tokio_postgres` error became a bare `RuntimeError`. Synapse's
transaction driver, though, branches on the *type* of a database error and on
its `pgcode`: `new_transaction` retries `OperationalError`, retries deadlocks it
recognises via `is_deadlock` (which reads `pgcode`) on a `DatabaseError`, and
`simple_upsert` retries `IntegrityError`. With everything collapsed to
`RuntimeError` none of that fired.

Add just the distinctions Synapse acts on, rather than psycopg2's full PEP-249
hierarchy: `Error` -> `DatabaseError` -> {`OperationalError`, `IntegrityError`},
exposed on the `postgres` submodule, each instance tagged with `pgcode` (the
SQLSTATE string, or `None`). A small classifier maps the SQLSTATE class:
constraint violations (`23`) to `IntegrityError`, connection/resource classes
(`08`/`53`/`57`/`58`) to `OperationalError`, everything else (incl. `40*`
deadlocks, which retry via `pgcode`) to `DatabaseError`. Codeless errors are
split with `is_closed()`: a lost connection is operational, any other (a bad
parameter, a failed connect) is a plain `DatabaseError` so it isn't retried.

Errors surfacing while a result stream is drained (the usual case for an
`INSERT` constraint violation) now route through the same mapping, so they carry
the right class and `pgcode` too.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/synapse_rust/test_database_postgres.py

@@ -110,9 +110,9 @@ class PostgresConnectionTestCase(unittest.TestCase):
     # ------------------------------------------------------------------
 
     def test_connect_bad_dsn_raises(self) -> None:
-        # A syntactically valid but unconnectable DSN should raise rather than
-        # return a half-open connection.
-        with self.assertRaises(RuntimeError):
+        # A syntactically valid but unconnectable DSN should raise one of our
+        # DBAPI2 errors rather than return a half-open connection.
+        with self.assertRaises(postgres.Error):
             _connect("host=127.0.0.1 port=1 dbname=does_not_exist")
 
     # ------------------------------------------------------------------
@@ -468,8 +468,12 @@ class PostgresConnectionTestCase(unittest.TestCase):
             cursor.execute("CREATE TEMP TABLE oor (x int4)")
             cursor.execute("INSERT INTO oor VALUES ($1)", [10**12])
 
-        with self.assertRaises(RuntimeError):
+        # This is a client-side encoding error (no SQLSTATE), so it surfaces as
+        # a plain DatabaseError -- not an OperationalError, since retrying a
+        # deterministic bad value would be pointless.
+        with self.assertRaises(postgres.DatabaseError) as ctx:
             run_interaction(self.conn, interaction)
+        self.assertNotIsInstance(ctx.exception, postgres.OperationalError)
 
     # ------------------------------------------------------------------
     # rowcount()
@@ -583,10 +587,10 @@ class PostgresConnectionTestCase(unittest.TestCase):
 
         def bad_sql(cursor: Any) -> None:
             # A syntax error: the server rejects this during prepare, which
-            # surfaces as a RuntimeError and rolls the transaction back.
+            # surfaces as a DatabaseError and rolls the transaction back.
             cursor.execute("SELECT FROM WHERE not valid sql")
 
-        with self.assertRaises(RuntimeError):
+        with self.assertRaises(postgres.DatabaseError):
             run_interaction(self.conn, bad_sql)
 
         # The transaction was rolled back and the connection handed back clean,
@@ -606,12 +610,15 @@ class PostgresConnectionTestCase(unittest.TestCase):
             cursor.execute("INSERT INTO uniq VALUES (1)")
             # Duplicate key. The error is reported by the server while the
             # statement's result stream is driven, so we drain it (via
-            # rowcount) to surface it as a RuntimeError.
+            # rowcount) to surface it -- as an IntegrityError, the same class
+            # (with the same 23505 pgcode) it would carry had it surfaced at
+            # execute time.
             cursor.execute("INSERT INTO uniq VALUES (1)")
             cursor.rowcount()
 
-        with self.assertRaises(RuntimeError):
+        with self.assertRaises(postgres.IntegrityError) as ctx:
             run_interaction(self.conn, violate)
+        self.assertEqual(ctx.exception.pgcode, "23505")
 
         # TEMP table lived only in the rolled-back transaction; the connection
         # itself is fine.
@@ -823,6 +830,71 @@ class PostgresConnectionDrivenTestCase(unittest.TestCase):
         self.conn.rollback()
 
 
+@unittest.skip_unless(
+    bool(USE_POSTGRES_FOR_TESTS), "requires a Postgres server (set SYNAPSE_POSTGRES)"
+)
+class PostgresErrorMappingTestCase(unittest.TestCase):
+    """The DBAPI2 exception hierarchy and the SQLSTATE→exception mapping.
+
+    Synapse's transaction driver branches on the *type* of the exception a
+    database call raises (``OperationalError`` → retry, ``IntegrityError`` →
+    retry upserts) and on its ``pgcode`` (``is_deadlock``). These tests check
+    the Rust backend raises the right class and carries a ``pgcode``, the way
+    psycopg2 does.
+    """
+
+    def setUp(self) -> None:
+        self.conn = postgres.connect(_build_dsn())
+
+    def tearDown(self) -> None:
+        del self.conn
+
+    def _exec_commit(self, sql: str) -> None:
+        """Run a single statement and commit it (its own transaction)."""
+        self.conn.cursor().execute(sql)
+        self.conn.commit()
+
+    # -- the hierarchy exposed on the module --------------------------------
+
+    def test_module_exposes_dbapi2_hierarchy(self) -> None:
+        """The exception attributes Synapse's engine code and DBAPI2Module
+        protocol rely on, with the expected subclass links."""
+        self.assertTrue(issubclass(postgres.DatabaseError, postgres.Error))
+        self.assertTrue(issubclass(postgres.OperationalError, postgres.DatabaseError))
+        self.assertTrue(issubclass(postgres.IntegrityError, postgres.DatabaseError))
+
+    # -- SQLSTATE → exception class -----------------------------------------
+
+    def test_unique_violation_is_integrity_error(self) -> None:
+        """A constraint violation raises ``IntegrityError`` with pgcode 23505."""
+        table = "rust_pg_err_integrity"
+        try:
+            self._exec_commit(f"CREATE TABLE {table} (id int PRIMARY KEY)")
+            self._exec_commit(f"INSERT INTO {table} VALUES (1)")
+
+            cursor = self.conn.cursor()
+            with self.assertRaises(postgres.IntegrityError) as ctx:
+                cursor.execute(f"INSERT INTO {table} VALUES (1)")
+                # The INSERT's error is reported while its result stream is
+                # driven, so drain it (via rowcount) to surface it.
+                cursor.rowcount()
+            self.assertEqual(ctx.exception.pgcode, "23505")
+            self.conn.rollback()
+        finally:
+            self._exec_commit(f"DROP TABLE IF EXISTS {table}")
+
+    def test_undefined_table_is_plain_database_error(self) -> None:
+        """An error we don't single out surfaces as a plain ``DatabaseError``
+        (not one of the specialised subclasses), still carrying its pgcode."""
+        cursor = self.conn.cursor()
+        with self.assertRaises(postgres.DatabaseError) as ctx:
+            cursor.execute("SELECT * FROM rust_pg_no_such_table")
+        self.assertNotIsInstance(ctx.exception, postgres.OperationalError)
+        self.assertNotIsInstance(ctx.exception, postgres.IntegrityError)
+        self.assertEqual(ctx.exception.pgcode, "42P01")  # undefined_table
+        self.conn.rollback()
+
+
 @unittest.skip_unless(
     bool(USE_POSTGRES_FOR_TESTS) and POSTGRES_HOST in (None, "", "localhost"),
     "requires Postgres reachable on libpq's default host",