Add runQuery / runOperation / connect to the Rust connection pool

`RustConnectionPool` stands in for `DatabasePool._db_pool`, a
`twisted.enterprise.adbapi.ConnectionPool`. It presented the slice Synapse's
`DatabasePool` uses (`runWithConnection`, `threadID`), but some tests reach for
`_db_pool` directly and call adbapi's `runQuery` / `runOperation` / `connect`,
which weren't implemented — so those tests errored with `AttributeError` on the
Rust backend.

Add them, mirroring adbapi's semantics:

  - `runQuery` / `runOperation` run one statement in its own transaction on a
    worker thread (commit on success, roll back on error), returning the rows
    (or nothing) as a `Deferred`.
  - `connect` returns a connection for direct synchronous use on the caller's
    thread, cached per thread as adbapi does and closed together with the pool.

Fixes the tests.storage.test_appservice and tests.storage.test_rollback_worker
failures on the Rust backend (26/26 pass); psycopg2 and sqlite are unaffected
(they use the real adbapi pool).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W3G4M92AmwSSZCbmtMJU3d

synapse/storage/rust_pool.py

@@ -22,10 +22,11 @@ result back to the reactor as a ``Deferred``.
 
 It presents the slice of ``twisted.enterprise.adbapi.ConnectionPool`` that
 ``DatabasePool`` uses — ``runWithConnection``, ``threadID``, ``threadpool`` and
-``running`` — so it can stand in for ``_db_pool`` (paired with
-:class:`~synapse.storage.engines.RustPostgresEngine`, which drives the wrapped
-connection). ``make_pool`` returns one of these when the database is configured
-with ``use_rust_driver``.
+``running`` — plus the ``runQuery`` / ``runOperation`` / ``connect`` convenience
+methods that some tests call on ``_db_pool`` directly, so it can stand in for it
+(paired with :class:`~synapse.storage.engines.RustPostgresEngine`, which drives
+the wrapped connection). ``make_pool`` returns one of these when the database is
+configured with ``use_rust_driver``.
 """
 
 import logging
@@ -101,6 +102,9 @@ class RustConnectionPool:
         )
         self.threadpool = ThreadPool(minthreads=1, maxthreads=threads, name=name)
         self.running = False
+        # Connections handed out by `connect()` for direct synchronous use, one
+        # cached per thread (adbapi's model). Closed together with the pool.
+        self._connections: dict[int, DBAPI2Connection] = {}
 
     def start(self) -> None:
         """Start the thread pool. Idempotent."""
@@ -121,6 +125,9 @@ class RustConnectionPool:
             return
         self.running = False
         self.threadpool.stop()
+        for conn in self._connections.values():
+            conn.close()
+        self._connections.clear()
         self._pool.close()
 
     def runWithConnection(  # noqa: N802 (implements adbapi's interface)
@@ -163,6 +170,77 @@ class RustConnectionPool:
         """
         return threading.get_ident()
 
+    def runQuery(  # noqa: N802 (implements adbapi's interface)
+        self, *args: Any, **kwargs: Any
+    ) -> "Deferred[list[Any]]":
+        """Run a single query in its own transaction and return all its rows.
+
+        Mirrors ``twisted.enterprise.adbapi.ConnectionPool.runQuery``: the
+        arguments are passed straight to the cursor's ``execute``, the
+        transaction is committed on success (rolled back on error), and the
+        result is the ``fetchall()``. Synapse itself goes through
+        ``DatabasePool``; this is here for callers that use ``_db_pool``
+        directly (e.g. tests).
+        """
+        return self.runWithConnection(self._run_query, args, kwargs)
+
+    def runOperation(  # noqa: N802 (implements adbapi's interface)
+        self, *args: Any, **kwargs: Any
+    ) -> "Deferred[None]":
+        """Run a single statement in its own transaction, discarding any rows.
+
+        Mirrors ``twisted.enterprise.adbapi.ConnectionPool.runOperation``.
+        """
+        return self.runWithConnection(self._run_operation, args, kwargs)
+
+    def connect(self) -> DBAPI2Connection:
+        """Return a connection for direct, synchronous use on the caller's thread.
+
+        Mirrors ``twisted.enterprise.adbapi.ConnectionPool.connect``: one
+        connection is cached per thread and reused (recreated if it has been
+        closed), and all cached connections are closed when the pool closes.
+        Unlike ``runWithConnection`` this does no thread hop — the caller drives
+        the connection itself, as schema-preparation code and some tests do.
+        """
+        tid = self.threadID()
+        conn = self._connections.get(tid)
+        if conn is None or conn.is_closed():
+            conn = DBAPI2Connection(self._pool.connect(), pool=self._pool)
+            self._connections[tid] = conn
+        return conn
+
+    def _run_query(
+        self, conn: DBAPI2Connection, args: tuple, kwargs: dict
+    ) -> list[Any]:
+        def body(cursor: Any) -> list[Any]:
+            cursor.execute(*args, **kwargs)
+            return cursor.fetchall()
+
+        return self._in_transaction(conn, body)
+
+    def _run_operation(self, conn: DBAPI2Connection, args: tuple, kwargs: dict) -> None:
+        def body(cursor: Any) -> None:
+            cursor.execute(*args, **kwargs)
+
+        self._in_transaction(conn, body)
+
+    @staticmethod
+    def _in_transaction(conn: DBAPI2Connection, body: Callable[[Any], R]) -> R:
+        """Run ``body(cursor)`` in a transaction: commit on success, else roll back.
+
+        Matches adbapi's ``_runInteraction`` ordering — the commit is part of the
+        protected region, so a failure to commit also rolls back.
+        """
+        cursor = conn.cursor()
+        try:
+            result = body(cursor)
+            cursor.close()
+            conn.commit()
+            return result
+        except Exception:
+            conn.rollback()
+            raise
+
     def _run(
         self,
         func: Callable[..., R],