LLM attempt 1

rust/src/handlers/mod.rs

@@ -24,13 +24,12 @@ use pyo3::{
 use crate::config::SynapseConfig;
 use crate::storage::db::python_db_pool::PythonDatabasePoolWrapper;
 use crate::storage::store::Store;
-use crate::UnwrapInfallible;
 
 pub mod versions;
 
 #[pyclass]
 struct RustHandlers {
-    versions: versions::VersionsHandler,
+    versions: Py<versions::VersionsHandler>,
 }
 
 #[pymethods]
@@ -40,14 +39,17 @@ impl RustHandlers {
     pub fn py_new(py: Python<'_>, homeserver: &Bound<'_, PyAny>) -> PyResult<RustHandlers> {
         let config: SynapseConfig = homeserver.getattr("config")?.extract()?;
 
+        // The Twisted reactor, used both to drive our Tokio runtime and to
+        // marshal database work back onto the reactor thread.
+        let reactor: Py<PyAny> = homeserver.call_method0("get_reactor")?.unbind();
+
         // hs.get_datastores().main.db_pool
-        let db_pool: PythonDatabasePoolWrapper = homeserver
+        let db_pool_py: Py<PyAny> = homeserver
             .call_method0("get_datastores")?
-            .into_pyobject(py)
-            .unwrap_infallible()
             .getattr("main")?
             .getattr("db_pool")?
-            .extract()?;
+            .unbind();
+        let db_pool = PythonDatabasePoolWrapper::new(db_pool_py, reactor.clone_ref(py));
 
         // Store is shared across all of the handlers so let's use an `Arc`
         let store = Arc::new(Store {
@@ -55,12 +57,21 @@ impl RustHandlers {
             db_pool: Box::new(db_pool),
         });
 
-        Ok(RustHandlers {
-            versions: versions::VersionsHandler {
+        let versions = Py::new(
+            py,
+            versions::VersionsHandler {
                 config: config.clone(),
                 store: Arc::clone(&store),
+                reactor: reactor.clone_ref(py),
             },
-        })
+        )?;
+
+        Ok(RustHandlers { versions })
+    }
+
+    #[getter]
+    fn versions(&self, py: Python<'_>) -> Py<versions::VersionsHandler> {
+        self.versions.clone_ref(py)
     }
 }
 

rust/src/handlers/versions.rs

@@ -13,10 +13,14 @@
  *
  */
 
-use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 
+use pyo3::prelude::*;
+use pythonize::{pythonize, PythonizeError};
+use serde::{Deserialize, Serialize};
+
 use crate::config::SynapseConfig;
+use crate::http_client::create_deferred;
 use crate::storage::store::{PerUserExperimentalFeature, Store};
 
 /// `GET /_matrix/client/versions` response
@@ -27,30 +31,73 @@ struct VersionsResponse {
     unstable_features: std::collections::BTreeMap<String, bool>,
 }
 
+impl<'py> IntoPyObject<'py> for VersionsResponse {
+    type Target = PyAny;
+    type Output = Bound<'py, Self::Target>;
+    type Error = PythonizeError;
+
+    fn into_pyobject(self, py: Python<'py>) -> Result<Self::Output, Self::Error> {
+        pythonize(py, &self)
+    }
+}
+
+#[pyclass]
 pub struct VersionsHandler {
     pub config: SynapseConfig,
     pub store: Arc<Store>,
+    /// The Twisted reactor, used to bridge our `async` response back into a
+    /// Twisted deferred that Python can `await`.
+    pub reactor: Py<PyAny>,
 }
 
+#[pymethods]
 impl VersionsHandler {
-    /// Assemble a `/versions` response
-    async fn get_versions(&self, user_id: Option<&str>) -> Result<VersionsResponse, anyhow::Error> {
+    /// Assemble a `/versions` response, returning a Twisted deferred that
+    /// resolves to the response body (a dict).
+    #[pyo3(signature = (user_id=None))]
+    fn get_versions<'py>(
+        &self,
+        py: Python<'py>,
+        user_id: Option<String>,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let store = Arc::clone(&self.store);
+        let config = self.config.clone();
+
+        create_deferred(py, self.reactor.bind(py), async move {
+            build_versions_response(&store, &config, user_id.as_deref())
+                .await
+                .map_err(|err| {
+                    pyo3::exceptions::PyRuntimeError::new_err(format!(
+                        "Failed to build /versions response: {err:#}"
+                    ))
+                })
+        })
+    }
+}
+
+/// Assemble a `/versions` response body.
+async fn build_versions_response(
+    store: &Store,
+    config: &SynapseConfig,
+    user_id: Option<&str>,
+) -> Result<VersionsResponse, anyhow::Error> {
+    {
         let msc3881_enabled = match user_id {
             Some(user_id) => {
-                self.store
+                store
                     .is_feature_enabled(user_id, PerUserExperimentalFeature::MSC3881)
                     .await?
             }
-            None => PerUserExperimentalFeature::MSC3881.is_globally_enabled(&self.config),
+            None => PerUserExperimentalFeature::MSC3881.is_globally_enabled(config),
         };
 
         let msc3575_enabled = match user_id {
             Some(user_id) => {
-                self.store
+                store
                     .is_feature_enabled(user_id, PerUserExperimentalFeature::MSC3575)
                     .await?
             }
-            None => PerUserExperimentalFeature::MSC3575.is_globally_enabled(&self.config),
+            None => PerUserExperimentalFeature::MSC3575.is_globally_enabled(config),
         };
 
         // TODO: Calculate these once since they shouldn't change after start-up.
@@ -67,7 +114,7 @@ impl VersionsHandler {
         //     in config.room.encryption_enabled_by_default_for_room_presets
         // );
 
-        return Ok(VersionsResponse {
+        Ok(VersionsResponse {
             versions: Vec::from([
                 // XXX: at some point we need to decide whether we need to include
                 // the previous version numbers, given we've defined r0.3.0 to be
@@ -171,6 +218,6 @@ impl VersionsHandler {
                 // // MSC4445: Sync timeline order
                 // ("org.matrix.msc4445.initial_sync_timeline_topological_ordering".to_string(), true),
             ]),
-        });
+        })
     }
 }

rust/src/http_client.rs

@@ -267,7 +267,7 @@ impl HttpClient {
 /// tokio runtime.
 ///
 /// Does not handle deferred cancellation or contextvars.
-fn create_deferred<'py, F, O>(
+pub(crate) fn create_deferred<'py, F, O>(
     py: Python<'py>,
     reactor: &Bound<'py, PyAny>,
     fut: F,

rust/src/storage/db/mod.rs

@@ -13,33 +13,113 @@
  *
  */
 
+use std::any::Any;
 use std::future::Future;
 
+use futures::future::BoxFuture;
+
 pub mod python_db_pool;
 pub mod rust_db_pool;
 
-// Using `Send + Sync` traits so this can stored in the `#[pyclass]` just fine
+/// A single database row, represented as the textual value of each column.
+///
+/// This is intentionally a lossy, engine-agnostic representation: it is the
+/// lowest common denominator that both the Python (`LoggingTransaction`) and
+/// native `tokio-postgres` backends can produce. Callers are responsible for
+/// parsing the strings into richer types as needed.
+pub type Row = Vec<String>;
+
+/// The type-erased result of a `run_interaction` callback.
+///
+/// We box the result as `dyn Any` so that the [`DatabasePool`] trait can stay
+/// object-safe (and therefore usable as `Box<dyn DatabasePool>`) while still
+/// allowing callbacks to return an arbitrary `R`. The ergonomic, generic
+/// [`DatabasePoolExt::run_interaction`] downcasts this back to the concrete
+/// type for the caller.
+pub type AnyResult = anyhow::Result<Box<dyn Any + Send>>;
+
+/// A type-erased `run_interaction` callback.
+///
+/// The callback is given a [`Transaction`] and returns a boxed future
+/// resolving to a type-erased result. It may be invoked multiple times under
+/// certain failure modes (serialization and deadlock errors), so it is `Fn`
+/// rather than `FnOnce`.
+pub type InteractionFn =
+    Box<dyn for<'txn> Fn(&'txn mut dyn Transaction) -> BoxFuture<'txn, AnyResult> + Send>;
+
+/// A database connection pool.
+///
+/// We use a `Box<dyn DatabasePool>` so the same code can run against either the
+/// Python-backed pool (in Synapse, see [`python_db_pool`]) or a native
+/// `tokio-postgres` pool (in `synapse-rust-apps`, see [`rust_db_pool`]). To keep
+/// the trait object-safe, the only required method is the type-erased
+/// [`Self::run_interaction_erased`]; prefer the generic
+/// [`DatabasePoolExt::run_interaction`] at call sites.
+///
+/// `Send + Sync` so it can be stored in a `#[pyclass]` and shared across threads.
 #[async_trait::async_trait]
 pub trait DatabasePool: Send + Sync {
-    /// Starts a transaction on the database and runs a given function
+    /// Starts a transaction on the database and runs the given (type-erased)
+    /// function.
     ///
-    /// The given `func` may be called multiple times under certain failure modes (like
-    /// serialization and deadlock errors).
-    fn run_interaction<'txn, R, F>(
-        &'txn self,
+    /// The given `func` may be called multiple times under certain failure
+    /// modes (like serialization and deadlock errors).
+    async fn run_interaction_erased(
+        &self,
+        name: &'static str,
+        func: InteractionFn,
+    ) -> AnyResult;
+}
+
+/// Ergonomic, generic extension to [`DatabasePool`].
+///
+/// This is automatically implemented for every `DatabasePool` (including
+/// `dyn DatabasePool`) via the blanket impl below, and provides the typed
+/// `run_interaction` that callers actually use. It lives in a separate trait
+/// (rather than on `DatabasePool` directly) because a generic method would make
+/// `DatabasePool` no longer object-safe.
+pub trait DatabasePoolExt: DatabasePool {
+    /// Starts a transaction on the database and runs the given function,
+    /// returning its result.
+    ///
+    /// The given `func` may be called multiple times under certain failure
+    /// modes (like serialization and deadlock errors).
+    fn run_interaction<R, F>(
+        &self,
         name: &'static str,
         func: F,
-    ) -> impl Future<Output = anyhow::Result<R>> + 'txn
+    ) -> impl Future<Output = anyhow::Result<R>> + Send
     where
-        R: Send + Sync + 'static,
-        F: for<'f> Fn(&'f mut dyn Transaction) -> BoxFuture<'f, anyhow::Result<R>> + Send + 'static;
+        R: Send + 'static,
+        F: for<'txn> Fn(&'txn mut dyn Transaction) -> BoxFuture<'txn, anyhow::Result<R>>
+            + Send
+            + Sync
+            + 'static,
+    {
+        // Erase the concrete return type `R` into `Box<dyn Any>` so we can call
+        // through the object-safe `run_interaction_erased`.
+        let erased: InteractionFn = Box::new(move |txn| {
+            let fut = func(txn);
+            Box::pin(async move {
+                let value = fut.await?;
+                Ok(Box::new(value) as Box<dyn Any + Send>)
+            })
+        });
+
+        async move {
+            let boxed = self.run_interaction_erased(name, erased).await?;
+            Ok(*boxed
+                .downcast::<R>()
+                .expect("run_interaction return type mismatch (this is a Synapse programming error)"))
+        }
+    }
 }
 
+impl<T: DatabasePool + ?Sized> DatabasePoolExt for T {}
+
 /// A [`tokio_postgres::Transaction`] looking thing that we can use on the Rust side to
 /// interact with the database
 #[async_trait::async_trait]
-pub trait Transaction {
-    async fn query(&self, sql: &str, args: &[&str]) -> Result<Vec<Row>, anyhow::Error>;
+pub trait Transaction: Send {
+    async fn query(&mut self, sql: &str, args: &[&str]) -> Result<Vec<Row>, anyhow::Error>;
 }
-
-pub type Row = Vec<String>;

rust/src/storage/db/python_db_pool.rs

@@ -17,9 +17,12 @@
 //!  - Database pool [`PythonDatabasePoolWrapper`] which allows you to start a...
 //!  - transaction [`LoggingTransactionWrapper`] and query the database
 
-use pyo3::{intern, prelude::*, types::PyCFunction, types::PyList};
+use std::sync::{Arc, Mutex};
 
-use crate::storage::db::{DatabaseConnection, DatabasePool, Row, Transaction};
+use pyo3::{exceptions::PyRuntimeError, intern, prelude::*, types::PyCFunction, types::PyList};
+use tokio::sync::oneshot;
+
+use crate::storage::db::{AnyResult, DatabasePool, InteractionFn, Row, Transaction};
 
 /// The database engines we support in the Python side of Synapse
 #[derive(Copy, Clone, Debug)]
@@ -40,50 +43,176 @@ impl DatabaseEngine {
 
 /// Wrapper for a `DatabasePool` from the Python side of Synapse.
 pub struct PythonDatabasePoolWrapper {
-    /// The underlying `DatabasePool`
+    /// The underlying Python `DatabasePool`
     database_pool_py: Py<PyAny>,
+
+    /// The Twisted reactor. We need this to marshal back onto the reactor thread
+    /// (via `callFromThread`) when starting transactions, since Twisted's thread
+    /// pool machinery must be driven from there.
+    reactor: Py<PyAny>,
 }
 
-impl<'a, 'py> FromPyObject<'a, 'py> for PythonDatabasePoolWrapper {
-    type Error = PyErr;
-
-    /// Extract from a Python `DatabasePool` passed as an argument.
-    fn extract(database_pool_py: Borrowed<'a, 'py, PyAny>) -> PyResult<Self> {
-        Ok(Self {
-            database_pool_py: database_pool_py.to_owned().unbind(),
-        })
+impl PythonDatabasePoolWrapper {
+    /// Build a wrapper around the Python `DatabasePool` (e.g.
+    /// `hs.get_datastores().main.db_pool`) and the Twisted `reactor`.
+    pub fn new(database_pool_py: Py<PyAny>, reactor: Py<PyAny>) -> Self {
+        Self {
+            database_pool_py,
+            reactor,
+        }
     }
 }
 
 #[async_trait::async_trait]
 impl DatabasePool for PythonDatabasePoolWrapper {
-    fn run_interaction<'txn, R, F>(
-        &'txn self,
-        name: &'static str,
-        func: F,
-    ) -> impl Future<Output = anyhow::Result<R>> + 'txn
-    where
-        R: Send + Sync + 'static,
-        F: for<'f> Fn(&'f mut dyn Transaction) -> BoxFuture<'f, anyhow::Result<R>> + Send + 'static,
-    {
-        Python::attach(|py| -> PyResult<Vec<Row>> {
-            let callback_func =
-                PyCFunction::new_closure(py, None, None, move |args, _| -> PyResult<Py<PyAny>> {
-                    // We found our `LoggingTransactionWrapper`
-                    let txn: LoggingTransactionWrapper = args.get_item(0)?;
-                    func(txn);
-                });
+    async fn run_interaction_erased(&self, name: &'static str, func: InteractionFn) -> AnyResult {
+        // We bridge the Python-side `runInteraction` (a coroutine, run on the
+        // Twisted reactor + thread pool) back into our `async` Rust world using a
+        // oneshot channel that resolves when the resulting deferred fires.
+        let (tx, rx) = oneshot::channel::<AnyResult>();
 
-            let execute_fn = self
-                .database_pool_py
+        // `runInteraction` calls `func` with a `LoggingTransaction` on a DB
+        // thread and expects a synchronous return value. Since we can't
+        // round-trip an arbitrary Rust `R` back out through Python, the callback
+        // stashes the result here and we pick it up once the deferred fires.
+        let result_slot: Arc<Mutex<Option<AnyResult>>> = Arc::new(Mutex::new(None));
+
+        Python::attach(|py| -> PyResult<()> {
+            // (1) The callback that Python's `runInteraction` invokes on a DB
+            // thread with a `LoggingTransaction`. We drive `func` to completion
+            // here. The Python query path is synchronous under the hood, so it's
+            // safe to block this dedicated DB thread until the future resolves.
+            let callback_slot = Arc::clone(&result_slot);
+            let callback = PyCFunction::new_closure(
+                py,
+                None,
+                None,
+                move |args, _kwargs| -> PyResult<Py<PyAny>> {
+                    let py = args.py();
+                    let txn_py = args.get_item(0)?;
+                    let mut txn = txn_py.extract::<LoggingTransactionWrapper>()?;
+
+                    let result = futures::executor::block_on(func(&mut txn));
+
+                    match result {
+                        Ok(value) => {
+                            *callback_slot.lock().unwrap() = Some(Ok(value));
+                            Ok(py.None())
+                        }
+                        Err(err) => {
+                            // Re-raise into Python so `runInteraction` rolls the
+                            // transaction back (and can apply its retry logic for
+                            // serialization/deadlock errors).
+                            let py_err = anyhow_to_pyerr(&err);
+                            *callback_slot.lock().unwrap() = Some(Err(err));
+                            Err(py_err)
+                        }
+                    }
+                },
+            )?
+            .unbind();
+
+            // The oneshot sender, shared between the success and error callbacks
+            // (only one of which ever fires).
+            let sender = Arc::new(Mutex::new(Some(tx)));
+
+            // (2a) Fired when the transaction succeeds: hand the stashed result
+            // back to the awaiting task.
+            let success_slot = Arc::clone(&result_slot);
+            let success_sender = Arc::clone(&sender);
+            let on_success = PyCFunction::new_closure(
+                py,
+                None,
+                None,
+                move |args, _kwargs| -> PyResult<Py<PyAny>> {
+                    let result = success_slot.lock().unwrap().take().unwrap_or_else(|| {
+                        Err(anyhow::anyhow!("run_interaction produced no result"))
+                    });
+                    if let Some(tx) = success_sender.lock().unwrap().take() {
+                        let _ = tx.send(result);
+                    }
+                    Ok(args.py().None())
+                },
+            )?
+            .unbind();
+
+            // (2b) Fired when the transaction fails. Prefer the original error
+            // captured in the callback (it carries the Rust context); otherwise
+            // fall back to the Twisted `Failure` text.
+            let error_slot = Arc::clone(&result_slot);
+            let error_sender = Arc::clone(&sender);
+            let on_error = PyCFunction::new_closure(
+                py,
+                None,
+                None,
+                move |args, _kwargs| -> PyResult<Py<PyAny>> {
+                    let result = error_slot.lock().unwrap().take().unwrap_or_else(|| {
+                        let description = args
+                            .get_item(0)
+                            .and_then(|failure| failure.str())
+                            .map(|s| s.to_string_lossy().into_owned())
+                            .unwrap_or_else(|_| "<unknown failure>".to_owned());
+                        Err(anyhow::anyhow!("run_interaction failed: {description}"))
+                    });
+                    if let Some(tx) = error_sender.lock().unwrap().take() {
+                        let _ = tx.send(result);
+                    }
+                    Ok(args.py().None())
+                },
+            )?
+            .unbind();
+
+            // (3) Kick off `runInteraction` on the reactor thread. It's a
+            // coroutine, so we wrap it with `ensureDeferred` and attach our
+            // callbacks.
+            let database_pool_py = self.database_pool_py.clone_ref(py);
+            let starter = PyCFunction::new_closure(
+                py,
+                None,
+                None,
+                move |args, _kwargs| -> PyResult<Py<PyAny>> {
+                    let py = args.py();
+                    let coro = database_pool_py.bind(py).call_method1(
+                        intern!(py, "runInteraction"),
+                        (name, callback.bind(py)),
+                    )?;
+                    let deferred = py
+                        .import("twisted.internet.defer")?
+                        .call_method1(intern!(py, "ensureDeferred"), (coro,))?;
+                    deferred.call_method1(
+                        intern!(py, "addCallbacks"),
+                        (on_success.bind(py), on_error.bind(py)),
+                    )?;
+                    Ok(py.None())
+                },
+            )?;
+
+            self.reactor
                 .bind(py)
-                .getattr(intern!(py, "runInteraction"))?;
-            let results = execute_fn.call1((callback_func,))?;
-            results
+                .call_method1(intern!(py, "callFromThread"), (starter,))?;
+
+            Ok(())
         })
+        .map_err(anyhow::Error::from)?;
+
+        rx.await
+            .map_err(|_| anyhow::anyhow!("run_interaction channel closed before completing"))?
     }
 }
 
+/// Convert an [`anyhow::Error`] into a [`PyErr`] to re-raise into Python.
+///
+/// If the error wraps an original Python exception (e.g. a database error
+/// surfaced through [`Transaction::query`]), we re-raise *that* exception so
+/// Synapse's transaction machinery can apply its retry logic
+/// (serialization/deadlock detection) on the real error.
+fn anyhow_to_pyerr(err: &anyhow::Error) -> PyErr {
+    if let Some(py_err) = err.downcast_ref::<PyErr>() {
+        return Python::attach(|py| py_err.clone_ref(py));
+    }
+    PyRuntimeError::new_err(format!("{err:#}"))
+}
+
 fn detect_engine(txn_py: &Bound<'_, PyAny>) -> PyResult<DatabaseEngine> {
     let name = txn_py
         .getattr("database_engine")
@@ -131,28 +260,8 @@ impl<'a, 'py> FromPyObject<'a, 'py> for LoggingTransactionWrapper {
     }
 }
 
-pub trait ValidDatabaseFieldType {}
-pub trait ValidDatabaseReturnType {}
-impl ValidDatabaseFieldType for String {}
-impl ValidDatabaseFieldType for usize {}
-impl<T: ValidDatabaseFieldType> ValidDatabaseFieldType for Option<T> {}
-impl<T0: ValidDatabaseFieldType> ValidDatabaseReturnType for (T0,) {}
-impl<T0: ValidDatabaseFieldType, T1: ValidDatabaseFieldType> ValidDatabaseReturnType for (T0, T1) {}
-impl<T0: ValidDatabaseFieldType, T1: ValidDatabaseFieldType, T2: ValidDatabaseFieldType>
-    ValidDatabaseReturnType for (T0, T1, T2)
-{
-}
-impl<
-        T0: ValidDatabaseFieldType,
-        T1: ValidDatabaseFieldType,
-        T2: ValidDatabaseFieldType,
-        T3: ValidDatabaseFieldType,
-    > ValidDatabaseReturnType for (T0, T1, T2, T3)
-{
-}
-
 impl LoggingTransactionWrapper {
-    pub fn execute<'py>(
+    fn execute<'py>(
         &mut self,
         py: Python<'py>,
         sql: &str,
@@ -165,30 +274,38 @@ impl LoggingTransactionWrapper {
         execute_fn.call1((sql, args))?;
         Ok(())
     }
-
-    pub fn fetchall<'py, T: FromPyObjectOwned<'py> + ValidDatabaseReturnType>(
-        &mut self,
-        py: Python<'py>,
-    ) -> anyhow::Result<Vec<T>> {
-        let fetch_fn = self
-            .logging_transaction_py
-            .bind(py)
-            .getattr(intern!(py, "fetchall"))?;
-        Ok(fetch_fn.call0()?.extract()?)
-    }
 }
 
 #[async_trait::async_trait]
 impl Transaction for LoggingTransactionWrapper {
-    async fn query(&self, sql: &str, args: &[&str]) -> Result<Vec<Row>, anyhow::Error> {
+    async fn query(&mut self, sql: &str, args: &[&str]) -> Result<Vec<Row>, anyhow::Error> {
         Python::attach(|py| -> PyResult<Vec<Row>> {
-            // Convert the Rust `&[&str]` of SQL parameters into a Python sequence so it
-            // can be passed through to the Python-side `execute`.
+            // Convert the Rust `&[&str]` of SQL parameters into a Python sequence
+            // so it can be passed through to the Python-side `execute`. Note that
+            // `LoggingTransaction.execute` converts `?` placeholders into the
+            // appropriate param style for the underlying engine, so we pass
+            // `?`-style SQL.
             let args = PyList::new(py, args)?;
-            // Run the query
             self.execute(py, sql, args.as_any())?;
-            // Get the results
-            let rows = self.fetchall(py)?;
+
+            // Pull the rows back out. Each cell is converted to its textual
+            // representation so we have a single engine-agnostic `Row` type;
+            // callers parse the strings into richer types as needed.
+            let rows_py = self
+                .logging_transaction_py
+                .bind(py)
+                .call_method0(intern!(py, "fetchall"))?;
+
+            let mut rows: Vec<Row> = Vec::new();
+            for row_py in rows_py.try_iter()? {
+                let row_py = row_py?;
+                let mut row: Row = Vec::new();
+                for cell in row_py.try_iter()? {
+                    let cell = cell?;
+                    row.push(cell.str()?.to_string_lossy().into_owned());
+                }
+                rows.push(row);
+            }
 
             Ok(rows)
         })

rust/src/storage/db/rust_db_pool.rs

@@ -17,10 +17,11 @@
 // interfaces are compatible with `tokio-postgres`.
 
 use anyhow::Context;
+use bb8_postgres::tokio_postgres::{self, types::ToSql, IsolationLevel};
 use bb8_postgres::PostgresConnectionManager;
 use postgres_native_tls::MakeTlsConnector;
 
-use crate::storage::db::{DatabaseConnection, DatabasePool, Row, Transaction};
+use crate::storage::db::{AnyResult, DatabasePool, InteractionFn, Row, Transaction};
 
 /// Native Rust database access backed by `tokio-postgres` (for use in synapse-rust-apps)
 pub struct RustDatabasePool {
@@ -29,50 +30,106 @@ pub struct RustDatabasePool {
 
 #[async_trait::async_trait]
 impl DatabasePool for RustDatabasePool {
-    async fn get_connection(&self) -> Result<Box<dyn DatabaseConnection>, anyhow::Error> {
-        let mut conn = self
-            .db_pool
-            .get()
-            // .instrument(tracing::info_span!("acquire database connection"))
-            .await
-            .context("Failed to acquire database connection")?;
+    async fn run_interaction_erased(&self, _name: &'static str, func: InteractionFn) -> AnyResult {
+        // Like Synapse's `runInteraction`, retry the whole transaction on
+        // serialization/deadlock errors (which can happen under repeatable-read).
+        loop {
+            let mut conn = self
+                .db_pool
+                .get()
+                .await
+                .context("Failed to acquire database connection")?;
 
-        Ok(Box::new(RustConnection { connection: conn }))
+            // Repeatable-read isolation level (like Synapse).
+            let txn = conn
+                .build_transaction()
+                .isolation_level(IsolationLevel::RepeatableRead)
+                .start()
+                .await
+                .context("Failed to start transaction")?;
+
+            let mut wrapper = TokioPostgresTransaction { txn };
+            match func(&mut wrapper).await {
+                Ok(value) => {
+                    wrapper
+                        .txn
+                        .commit()
+                        .await
+                        .context("Failed to commit transaction")?;
+                    return Ok(value);
+                }
+                Err(err) => {
+                    // The transaction is rolled back implicitly when dropped, but
+                    // be explicit about it before deciding whether to retry.
+                    let _ = wrapper.txn.rollback().await;
+                    if is_retryable(&err) {
+                        continue;
+                    }
+                    return Err(err);
+                }
+            }
+        }
     }
 }
 
-pub struct RustConnection<'a> {
-    connection: bb8::PooledConnection<'a, PostgresConnectionManager<MakeTlsConnector>>,
-}
-
-impl DatabaseConnection for RustConnection<'_> {
-    async fn get_transaction(
-        &self,
-        _description: &str,
-    ) -> Result<Box<dyn Transaction>, anyhow::Error> {
-        // TODO: Set repeatable-read isolation level (like Synapse)
-        let txn = self
-            .connection
-            .transaction()
-            // .instrument(tracing::info_span!("start transaction"))
-            .await
-            .context("Failed to start transaction")?;
-
-        Ok(Box::new(TokioPostgresTransaction { txn }))
-    }
+/// Whether a failed transaction should be retried (serialization/deadlock errors).
+fn is_retryable(err: &anyhow::Error) -> bool {
+    err.downcast_ref::<tokio_postgres::Error>()
+        .and_then(|e| e.code())
+        .map(|code| {
+            *code == tokio_postgres::error::SqlState::T_R_SERIALIZATION_FAILURE
+                || *code == tokio_postgres::error::SqlState::T_R_DEADLOCK_DETECTED
+        })
+        .unwrap_or(false)
 }
 
 struct TokioPostgresTransaction<'a> {
-    txn: bb8_postgres::tokio_postgres::Transaction<'a>,
+    txn: tokio_postgres::Transaction<'a>,
 }
 
 #[async_trait::async_trait]
 impl Transaction for TokioPostgresTransaction<'_> {
-    async fn query(&self, sql: &str, args: &[&str]) -> Result<Vec<Row>, anyhow::Error> {
-        // TODO: Convert `?` SQL param style to `tokio-postgres` compatible
+    async fn query(&mut self, sql: &str, args: &[&str]) -> Result<Vec<Row>, anyhow::Error> {
+        // Synapse SQL uses `?` placeholders; `tokio-postgres` uses `$1`, `$2`, ...
+        let sql = convert_param_style(sql);
 
-        let rows = self.txn.query(sql, args).await?;
+        let params: Vec<&(dyn ToSql + Sync)> =
+            args.iter().map(|arg| arg as &(dyn ToSql + Sync)).collect();
+        let rows = self
+            .txn
+            .query(&sql, &params)
+            .await
+            .context("Failed to run query")?;
 
-        Ok(rows)
+        let mut out: Vec<Row> = Vec::with_capacity(rows.len());
+        for row in rows {
+            let mut cells: Row = Vec::with_capacity(row.len());
+            for i in 0..row.len() {
+                // Best-effort textual extraction to match the engine-agnostic
+                // `Row` type. A real implementation would map column types
+                // properly; this only exists to prove the interface fits.
+                let value: String = row.try_get(i).unwrap_or_default();
+                cells.push(value);
+            }
+            out.push(cells);
+        }
+
+        Ok(out)
     }
 }
+
+/// Convert `?`-style placeholders into `tokio-postgres`'s `$1`, `$2`, ... style.
+fn convert_param_style(sql: &str) -> String {
+    let mut out = String::with_capacity(sql.len());
+    let mut n = 0;
+    for ch in sql.chars() {
+        if ch == '?' {
+            n += 1;
+            out.push('$');
+            out.push_str(&n.to_string());
+        } else {
+            out.push(ch);
+        }
+    }
+    out
+}

rust/src/storage/store.rs

@@ -13,12 +13,16 @@
  *
  */
 
-use serde::{Serialize};
+use futures::FutureExt;
+use serde::Serialize;
 
-use crate::{config::SynapseConfig, storage::db::DatabasePool};
+use crate::{
+    config::SynapseConfig,
+    storage::db::{DatabasePool, DatabasePoolExt},
+};
 
 /// Currently supported per-user features
-#[derive(Serialize)]
+#[derive(Serialize, Debug)]
 pub enum PerUserExperimentalFeature {
     #[serde(rename = "msc3881")]
     MSC3881,
@@ -71,30 +75,48 @@ impl Store {
             return Ok(true);
         }
 
-        let is_feature_enabled_for_user = self.db_pool.run_interaction<bool>("is_feature_enabled_for_user", |txn| {
-            async move {
-                let rows = txn
-                    .query(
-                        r#"
-                        SELECT enabled
-                        FROM per_user_experimental_features
-                        WHERE user_id = ? AND feature = ?
-                        "#,
-                        &[user_id, &feature.to_string()],
-                    )
-                    .await;
+        // It's not enabled globally, so check whether it's enabled per-user.
+        //
+        // Owned copies so the callback can be `'static` (it may be moved to
+        // another thread and called multiple times under retries).
+        let user_id = user_id.to_string();
+        let feature = feature.to_string();
 
-                match (rows.len(), rows.first()) {
-                    (1, Some(enabled)) => enabled,
-                    (0, None) => false,
-                    _ => {
-                        panic!("Synapse programming error");
-                    }
+        let is_feature_enabled_for_user = self
+            .db_pool
+            .run_interaction("is_feature_enabled_for_user", move |txn| {
+                let user_id = user_id.clone();
+                let feature = feature.clone();
+                async move {
+                    let rows = txn
+                        .query(
+                            "SELECT enabled \
+                             FROM per_user_experimental_features \
+                             WHERE user_id = ? AND feature = ?",
+                            &[user_id.as_str(), feature.as_str()],
+                        )
+                        .await?;
+
+                    // `None` (no row) and a falsy value are treated the same.
+                    let enabled = rows
+                        .first()
+                        .and_then(|row| row.first())
+                        .is_some_and(|value| parse_db_bool(value));
+
+                    Ok(enabled)
                 }
-            }
-            .boxed()
-        }).await;
-        
+                .boxed()
+            })
+            .await?;
+
         Ok(is_feature_enabled_for_user)
     }
 }
+
+/// Parse a boolean as returned by either database engine.
+///
+/// Postgres renders `BOOLEAN` columns as `"True"`/`"False"` while SQLite stores
+/// them as integers (`"1"`/`"0"`).
+fn parse_db_bool(value: &str) -> bool {
+    matches!(value, "True" | "true" | "t" | "1")
+}

synapse/rest/client/versions.py

@@ -28,7 +28,6 @@ from typing import TYPE_CHECKING
 from synapse.http.server import HttpServer
 from synapse.http.servlet import RestServlet
 from synapse.http.site import SynapseRequest
-from synapse.synapse_rust.handlers.versions import get_versions
 from synapse.types import JsonDict
 
 if TYPE_CHECKING:
@@ -79,9 +78,7 @@ class VersionsRestServlet(RestServlet):
         # authenticated responses are not served from cache.
         request.setHeader(b"Vary", b"Authorization")
 
-        versions_response_body = await self.rust_handlers.versions.get_versions(
-            user_id, self.config
-        )
+        versions_response_body = await self.rust_handlers.versions.get_versions(user_id)
 
         return (
             200,

synapse/server.py

@@ -174,6 +174,7 @@ from synapse.state import StateHandler, StateResolutionHandler
 from synapse.storage import Databases
 from synapse.storage.controllers import StorageControllers
 from synapse.streams.events import EventSources
+from synapse.synapse_rust.handlers import RustHandlers
 from synapse.synapse_rust.msc4388_rendezvous import MSC4388RendezvousHandler
 from synapse.synapse_rust.rendezvous import RendezvousHandler
 from synapse.types import DomainSpecificString, ISynapseReactor

synapse/synapse_rust/handlers.pyi

@@ -0,0 +1,35 @@
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+from typing import TYPE_CHECKING, Optional
+
+from twisted.internet.defer import Deferred
+
+from synapse.types import JsonDict
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+class VersionsHandler:
+    def get_versions(self, user_id: Optional[str] = None) -> Deferred[JsonDict]:
+        """
+        Assemble a `/versions` response.
+
+        The returned deferred follows Synapse logcontext rules.
+        """
+
+class RustHandlers:
+    """The collection of Rust-implemented request handlers."""
+
+    def __init__(self, homeserver: "HomeServer") -> None: ...
+    @property
+    def versions(self) -> VersionsHandler: ...