From 8f0768d491083c12bb0b2bc0b9a5aa8856e8bb78 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Tue, 9 Jun 2026 20:22:29 -0500 Subject: [PATCH] LLM attempt 1 --- rust/src/handlers/mod.rs | 29 ++- rust/src/handlers/versions.rs | 65 ++++++- rust/src/http_client.rs | 2 +- rust/src/storage/db/mod.rs | 106 +++++++++-- rust/src/storage/db/python_db_pool.rs | 257 +++++++++++++++++++------- rust/src/storage/db/rust_db_pool.rs | 123 ++++++++---- rust/src/storage/store.rs | 72 +++++--- synapse/rest/client/versions.py | 5 +- synapse/server.py | 1 + synapse/synapse_rust/handlers.pyi | 35 ++++ 10 files changed, 531 insertions(+), 164 deletions(-) create mode 100644 synapse/synapse_rust/handlers.pyi diff --git a/rust/src/handlers/mod.rs b/rust/src/handlers/mod.rs index cc650a9d65..e42952ec1a 100644 --- a/rust/src/handlers/mod.rs +++ b/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, } #[pymethods] @@ -40,14 +39,17 @@ impl RustHandlers { pub fn py_new(py: Python<'_>, homeserver: &Bound<'_, PyAny>) -> PyResult { 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 = homeserver.call_method0("get_reactor")?.unbind(); + // hs.get_datastores().main.db_pool - let db_pool: PythonDatabasePoolWrapper = homeserver + let db_pool_py: Py = 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 { + self.versions.clone_ref(py) } } diff --git a/rust/src/handlers/versions.rs b/rust/src/handlers/versions.rs index 322d3a0620..d929759939 100644 --- a/rust/src/handlers/versions.rs +++ b/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, } +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 { + pythonize(py, &self) + } +} + +#[pyclass] pub struct VersionsHandler { pub config: SynapseConfig, pub store: Arc, + /// The Twisted reactor, used to bridge our `async` response back into a + /// Twisted deferred that Python can `await`. + pub reactor: Py, } +#[pymethods] impl VersionsHandler { - /// Assemble a `/versions` response - async fn get_versions(&self, user_id: Option<&str>) -> Result { + /// 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, + ) -> PyResult> { + 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 { + { 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), ]), - }); + }) } } diff --git a/rust/src/http_client.rs b/rust/src/http_client.rs index 398ba9041f..ecd5c7b55e 100644 --- a/rust/src/http_client.rs +++ b/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, diff --git a/rust/src/storage/db/mod.rs b/rust/src/storage/db/mod.rs index 8555ec0fb7..804c8ac937 100644 --- a/rust/src/storage/db/mod.rs +++ b/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; + +/// 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`) 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>; + +/// 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 Fn(&'txn mut dyn Transaction) -> BoxFuture<'txn, AnyResult> + Send>; + +/// A database connection pool. +/// +/// We use a `Box` 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( + &self, name: &'static str, func: F, - ) -> impl Future> + 'txn + ) -> impl Future> + Send where - R: Send + Sync + 'static, - F: for<'f> Fn(&'f mut dyn Transaction) -> BoxFuture<'f, anyhow::Result> + Send + 'static; + R: Send + 'static, + F: for<'txn> Fn(&'txn mut dyn Transaction) -> BoxFuture<'txn, anyhow::Result> + + Send + + Sync + + 'static, + { + // Erase the concrete return type `R` into `Box` 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) + }) + }); + + async move { + let boxed = self.run_interaction_erased(name, erased).await?; + Ok(*boxed + .downcast::() + .expect("run_interaction return type mismatch (this is a Synapse programming error)")) + } + } } +impl 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, anyhow::Error>; +pub trait Transaction: Send { + async fn query(&mut self, sql: &str, args: &[&str]) -> Result, anyhow::Error>; } - -pub type Row = Vec; diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index 0c9059ab3a..1939fe3749 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/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, + + /// 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, } -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 { - 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, reactor: Py) -> 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> + 'txn - where - R: Send + Sync + 'static, - F: for<'f> Fn(&'f mut dyn Transaction) -> BoxFuture<'f, anyhow::Result> + Send + 'static, - { - Python::attach(|py| -> PyResult> { - let callback_func = - PyCFunction::new_closure(py, None, None, move |args, _| -> PyResult> { - // 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::(); - 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>> = 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> { + let py = args.py(); + let txn_py = args.get_item(0)?; + let mut txn = txn_py.extract::()?; + + 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> { + 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> { + 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(|_| "".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> { + 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::() { + return Python::attach(|py| py_err.clone_ref(py)); + } + PyRuntimeError::new_err(format!("{err:#}")) +} + fn detect_engine(txn_py: &Bound<'_, PyAny>) -> PyResult { 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 ValidDatabaseFieldType for Option {} -impl ValidDatabaseReturnType for (T0,) {} -impl ValidDatabaseReturnType for (T0, T1) {} -impl - 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> { - 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, anyhow::Error> { + async fn query(&mut self, sql: &str, args: &[&str]) -> Result, anyhow::Error> { Python::attach(|py| -> PyResult> { - // 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 = 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) }) diff --git a/rust/src/storage/db/rust_db_pool.rs b/rust/src/storage/db/rust_db_pool.rs index 79a54be20e..addddd489d 100644 --- a/rust/src/storage/db/rust_db_pool.rs +++ b/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, 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>, -} - -impl DatabaseConnection for RustConnection<'_> { - async fn get_transaction( - &self, - _description: &str, - ) -> Result, 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::() + .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, anyhow::Error> { - // TODO: Convert `?` SQL param style to `tokio-postgres` compatible + async fn query(&mut self, sql: &str, args: &[&str]) -> Result, 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, ¶ms) + .await + .context("Failed to run query")?; - Ok(rows) + let mut out: Vec = 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 +} diff --git a/rust/src/storage/store.rs b/rust/src/storage/store.rs index a4929c435f..76d092efc0 100644 --- a/rust/src/storage/store.rs +++ b/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("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") +} diff --git a/synapse/rest/client/versions.py b/synapse/rest/client/versions.py index d8b0a0545d..9312a616da 100644 --- a/synapse/rest/client/versions.py +++ b/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, diff --git a/synapse/server.py b/synapse/server.py index f46f5845e4..635e444eb6 100644 --- a/synapse/server.py +++ b/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 diff --git a/synapse/synapse_rust/handlers.pyi b/synapse/synapse_rust/handlers.pyi new file mode 100644 index 0000000000..4c4f71b7bd --- /dev/null +++ b/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: +# . + +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: ...