Tidy edge cases in the Rust connection-pool adapter

Small robustness fixes to `RustConnectionPool`, independent of any query path:
guard `connectionFactory` against a closed pool (a clear error rather than an
`AttributeError` on `None`), snapshot the per-thread connections before closing
them so a concurrent `connect()` can't mutate the dict mid-iteration, drop the
now-unused `defer_to_threadpool` import, and correct the class docstring to
describe the raw-`Deferred` contract.

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

@@ -39,7 +39,6 @@ from typing_extensions import Concatenate, ParamSpec
 from twisted.internet import threads
 from twisted.python.threadpool import ThreadPool
 
-from synapse.logging.context import defer_to_threadpool
 from synapse.storage.rust_dbapi import Connection as DBAPI2Connection
 from synapse.synapse_rust.database import postgres
 
@@ -59,10 +58,11 @@ class RustConnectionPool:
 
     Each :meth:`runWithConnection` call runs its function on a worker thread
     with a (DBAPI2-adapter) connection checked out of the native Rust pool, and
-    returns a ``Deferred`` that fires on the reactor thread with the result (or
-    an errback if it raised). Log contexts are preserved across the hop,
-    following the same rules as
-    :func:`synapse.logging.context.defer_to_threadpool`.
+    returns a *raw* ``Deferred`` that fires on the reactor thread with the
+    result (or an errback if it raised) — like
+    ``adbapi.ConnectionPool.runWithConnection``, logcontext handling is the
+    caller's job (``DatabasePool.runWithConnection`` supplies the single
+    ``make_deferred_yieldable``; see :meth:`runWithConnection`).
     """
 
     def __init__(
@@ -124,7 +124,12 @@ class RustConnectionPool:
         # Check a connection out of the native pool and wrap it in the DBAPI2
         # adapter. `owns_pool=False`: the pool is shared and outlives the
         # checkout. Mirrors `adbapi.ConnectionPool.connectionFactory`.
-        return DBAPI2Connection(self._pool.connect(), pool=self._pool)
+        pool = self._pool
+        if pool is None:
+            # `close()` dropped the native pool and no `start()` has reopened
+            # it; a clear error beats an AttributeError on None.
+            raise RuntimeError("connection pool has been closed")
+        return DBAPI2Connection(pool.connect(), pool=pool)
 
     def start(self) -> None:
         """Start the thread pool. Idempotent.
@@ -152,7 +157,9 @@ class RustConnectionPool:
             return
         self.running = False
         self.threadpool.stop()
-        for conn in self._connections.values():
+        # Snapshot: a concurrent `connect()` on another thread must not change
+        # the dict's size under this iteration.
+        for conn in list(self._connections.values()):
             conn.close()
         self._connections.clear()
         if self._pool is not None: