From 9714a9468f8c1341a9f5d02257c7b024006a92af Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 24 Jun 2026 18:29:07 -0500 Subject: [PATCH 01/32] Add Rust database interface --- Cargo.lock | 12 ++ rust/Cargo.toml | 1 + rust/src/lib.rs | 1 + rust/src/storage/db/mod.rs | 250 +++++++++++++++++++++++++++++++++++++ rust/src/storage/mod.rs | 16 +++ 5 files changed, 280 insertions(+) create mode 100644 rust/src/storage/db/mod.rs create mode 100644 rust/src/storage/mod.rs diff --git a/Cargo.lock b/Cargo.lock index a6d98135cd7..d1daa638a6f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -23,6 +23,17 @@ version = "1.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457" +[[package]] +name = "async-trait" +version = "0.1.89" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -1319,6 +1330,7 @@ name = "synapse" version = "0.1.0" dependencies = [ "anyhow", + "async-trait", "base64", "blake2", "bytes", diff --git a/rust/Cargo.toml b/rust/Cargo.toml index 162bc981820..612ab09f6d0 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -22,6 +22,7 @@ crate-type = ["lib", "cdylib"] name = "synapse.synapse_rust" [dependencies] +async-trait = "0.1.89" anyhow = "1.0.63" base64 = "0.22.1" bytes = "1.6.0" diff --git a/rust/src/lib.rs b/rust/src/lib.rs index eb2d8cecdde..6e3ee06f75d 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -20,6 +20,7 @@ pub mod push; pub mod rendezvous; pub mod room_versions; pub mod segmenter; +pub mod storage; pub mod tokio_runtime; pub mod types; diff --git a/rust/src/storage/db/mod.rs b/rust/src/storage/db/mod.rs new file mode 100644 index 00000000000..a853eca6309 --- /dev/null +++ b/rust/src/storage/db/mod.rs @@ -0,0 +1,250 @@ +/* + * 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: + * . + * + */ + +use std::any::Any; +use std::future::Future; + +use futures::future::BoxFuture; +use futures::FutureExt; + +/// A type-erased `run_interaction` callback. +/// +/// This is the dyn-compatible form of the `func` passed to +/// [`DatabasePoolExt::run_interaction`]. +/// +/// The ergonomic [`DatabasePoolExt::run_interaction`] handles the boxing and downcasts +/// the result back to `R` for the caller. +/// +/// It may be invoked multiple times under certain failure modes (serialization +/// and deadlock errors), so it is `Fn` rather than `FnOnce`. +pub type ErasedInteraction = + Box Fn(&'txn mut dyn Transaction) -> BoxFuture<'txn, ErasedResult> + Send>; + +/// The type-erased *result* of an [`ErasedInteraction`] +/// [`DatabasePool::run_interaction_erased`]. +pub type ErasedResult = anyhow::Result>; + +/// A database connection pool. +/// +/// Held behind a trait object (e.g. `Box`) as it can be backed by +/// either the Python-backed pool (in Synapse, see [`python_db_pool`]) or a Rust native +/// `tokio-postgres` pool (expected to be used in `synapse-rust-apps`). +/// +/// To keep the trait dyn-compatible, we have to specify a type-erased +/// [`run_interaction_erased`](Self::run_interaction_erased) version; callers should +/// prefer the ergonomic, generic [`run_interaction`](DatabasePoolExt::run_interaction). +/// +/// `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 the given (type-erased) + /// `func`, returning its boxed result. + /// + /// Implementors implement this; callers should prefer + /// [`DatabasePoolExt::run_interaction`], which boxes up the result and + /// downcasts it back to the concrete type for you. + async fn run_interaction_erased( + &self, + name: &'static str, + func: ErasedInteraction, + ) -> ErasedResult; +} + +/// Ergonomic, strongly-typed access to a [`DatabasePool`]. +pub trait DatabasePoolExt: DatabasePool { + /// Starts a transaction on the database and runs the given function, + /// returning its result. + /// + /// `name` should be a descriptive identifier for logging/metrics + /// + /// `func` may be called multiple times under certain failure modes (like + /// serialization and deadlock errors), so it is `Fn` rather than `FnOnce`. + /// + /// `func` is async but you should only call `.await` on [`Transaction`] methods. + /// This is a minor cosmetic flaw but seems fine, as you don't want to be doing any + /// unnecessary waiting in your transaction anyway. + /// + /// Usage: + /// ```rust + /// db_pool + /// .run_interaction(|txn| { + /// async move { + /// /* do stuff with txn */ + /// } + /// .boxed() + /// }) + /// ``` + // + // Ideally, this method signature would be slightly different to allow downstream + // usage to look like the following (simpler) but because we allow the work to + // happen on other threads, the `Future` needs to be `Send`; As of 2026-06-22, the + // `AsyncFn` trait has no stable way to express that "the future this async closure + // produces is `Send`". The intended fix is probably return-type-notation + // (https://github.com/rust-lang/rust/issues/109417). + // ``` + // db_pool.run_interaction("description", async move |txn| { + // /* do stuff with txn */ + // }) + // ``` + // + // Refs: + // - [RFC 3668: Async closures](https://github.com/rust-lang/rfcs/pull/3668) + // - [RFC 3654: Return Type Notation](https://github.com/rust-lang/rfcs/pull/3654) + // - [Tracking Issue for return type notation](https://github.com/rust-lang/rust/issues/109417) + fn run_interaction( + &self, + name: &'static str, + func: F, + ) -> impl Future> + Send + where + R: Send + 'static, + F: for<'txn> Fn(&'txn mut dyn Transaction) -> BoxFuture<'txn, anyhow::Result> + + Send + + 'static, + { + // Erase the concrete return type `R` into `Box` so we can call + // through the dyn-compatible `run_interaction_erased`. + let erased: ErasedInteraction = Box::new(move |txn| { + let fut = func(txn); + async move { Ok(Box::new(fut.await?) as Box) }.boxed() + }); + + 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)", + )) + } + } +} + +/// Blanket-implemented for every [`DatabasePool`] so +/// [`run_interaction`](DatabasePoolExt::run_interaction) is always available +impl DatabasePoolExt for T {} + +/// A transaction to interact with the database +/// +/// Based on the ergonomics of [`tokio_postgres::Transaction`] +#[async_trait::async_trait] +pub trait Transaction: Send { + /// Run a database query, returning a list of resulting rows. + /// + /// We expect the `sql` query should use `?` placeholders for the `args`. Downstream + /// implementations should string-replace `?` as necessary. + // + // `async` as this is representing a round-trip between the app and database + async fn query(&mut self, sql: &str, args: &[&str]) -> Result, anyhow::Error>; +} + +/// A single backend-agnostic value within a [`DbRow`]. +/// +/// Each pool maps the values its database driver hands back into this common +/// set, so callers can work with one representation regardless of engine. +#[derive(Debug, Clone, PartialEq)] +pub enum DbValue { + /// A SQL `NULL`. + Null, + Bool(bool), + Int(i64), + Float(f64), + Text(String), +} + +/// A row of data returned from the database by a query. +/// +/// Each pool converts the cells its database driver hands back into the +/// engine-agnostic [`DbValue`] representation, so a row is simply a list of them. +/// Values are pulled out by their numeric index with [`DbRowExt::try_get`]. +pub type DbRow = Vec; + +/// Extension methods for reading typed values out of a [`DbRow`]. +/// +/// Based on [`tokio_postgres::Row`]'s `try_get`: [`try_get`](Self::try_get) +/// converts the [`DbValue`] at a given index into the requested type via +/// [`FromDbValue`] (our analogue of `tokio-postgres`'s `FromSql`). +pub trait DbRowExt { + /// Deserializes a value from the row, specified by its numeric index, + /// returning an error if the index is out of bounds or the value cannot be + /// converted into `T`. + fn try_get(&self, index: usize) -> Result; +} + +impl DbRowExt for DbRow { + fn try_get(&self, index: usize) -> Result { + let value = self.get(index).cloned().ok_or_else(|| { + anyhow::anyhow!( + "tried to get column {index} but the row only has {} column(s)", + self.len() + ) + })?; + + T::from_value(value) + } +} + +/// Converts a backend-agnostic [`DbValue`] into a concrete Rust type, analogous to +/// `tokio-postgres`'s `FromSql`. +pub trait FromDbValue: Sized { + fn from_value(value: DbValue) -> Result; +} + +impl FromDbValue for bool { + fn from_value(value: DbValue) -> Result { + match value { + DbValue::Bool(b) => Ok(b), + // SQLite has no native boolean type and stores them as integers. + DbValue::Int(i) => Ok(i != 0), + other => anyhow::bail!("cannot read {other:?} as bool"), + } + } +} + +impl FromDbValue for i64 { + fn from_value(value: DbValue) -> Result { + match value { + DbValue::Int(i) => Ok(i), + DbValue::Bool(b) => Ok(b as i64), + other => anyhow::bail!("cannot read {other:?} as i64"), + } + } +} + +impl FromDbValue for f64 { + fn from_value(value: DbValue) -> Result { + match value { + DbValue::Float(f) => Ok(f), + DbValue::Int(i) => Ok(i as f64), + other => anyhow::bail!("cannot read {other:?} as f64"), + } + } +} + +impl FromDbValue for String { + fn from_value(value: DbValue) -> Result { + match value { + DbValue::Text(s) => Ok(s), + other => anyhow::bail!("cannot read {other:?} as String"), + } + } +} + +impl FromDbValue for Option { + fn from_value(value: DbValue) -> Result { + match value { + DbValue::Null => Ok(None), + other => Ok(Some(T::from_value(other)?)), + } + } +} diff --git a/rust/src/storage/mod.rs b/rust/src/storage/mod.rs new file mode 100644 index 00000000000..ac2c181452f --- /dev/null +++ b/rust/src/storage/mod.rs @@ -0,0 +1,16 @@ +/* + * 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: + * . + * + */ + +pub mod db; From 6da263b6c733d5426243c241a64fcd00f1dba12c Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 24 Jun 2026 18:34:26 -0500 Subject: [PATCH 02/32] Add `run_python_awaitable(...)` --- rust/src/deferred.rs | 168 ++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 158 insertions(+), 10 deletions(-) diff --git a/rust/src/deferred.rs b/rust/src/deferred.rs index 91d5a08f7db..b915d65e078 100644 --- a/rust/src/deferred.rs +++ b/rust/src/deferred.rs @@ -13,10 +13,17 @@ * */ -use std::{future::Future, sync::OnceLock}; +use std::{ + future::Future, + sync::{Arc, Mutex}, +}; use once_cell::sync::OnceCell; -use pyo3::{create_exception, exceptions::PyException, prelude::*}; +use pyo3::{ + create_exception, exceptions::PyException, exceptions::PyRuntimeError, intern, prelude::*, + types::PyCFunction, +}; +use tokio::sync::oneshot; use crate::tokio_runtime::runtime; @@ -51,10 +58,18 @@ fn defer(py: Python<'_>) -> PyResult<&Bound<'_, PyAny>> { .bind(py)) } +/// A reference to the `synapse.logging.context` module. +static LOGGING_CONTEXT_MODULE: OnceCell> = OnceCell::new(); + +/// Access to the `synapse.logging.context` module. +fn logging_context_module(py: Python<'_>) -> PyResult<&Bound<'_, PyAny>> { + Ok(LOGGING_CONTEXT_MODULE + .get_or_try_init(|| py.import("synapse.logging.context").map(Into::into))? + .bind(py)) +} + /// Creates a twisted deferred from the given future, spawning the task on the /// tokio runtime. -/// -/// Does not handle deferred cancellation or contextvars. pub fn create_deferred<'py, F, O>( py: Python<'py>, reactor: &Bound<'py, PyAny>, @@ -110,18 +125,149 @@ where make_deferred_yieldable(py, &deferred) } -static MAKE_DEFERRED_YIELDABLE: OnceLock> = OnceLock::new(); +/// Runs a Python awaitable to completion on the Twisted reactor and resolves +/// with its result. +/// +/// This is the inverse of [`create_deferred`]: where that turns a Rust future +/// into a Twisted `Deferred`, this turns a Python awaitable into a Rust future. +/// +/// Despite returning a future, the awaitable is kicked off in the background running in +/// the Twisted reactor and runs to completion regardless of whether the returned Rust +/// future is ever polled; awaiting it only observes the result. +pub(crate) async fn run_python_awaitable( + reactor: Py, + make_awaitable: F, +) -> PyResult> +where + F: for<'py> Fn(Python<'py>) -> PyResult> + Send + 'static, +{ + // Resolves when the awaitable completes; carries the resolved value or error. + let (tx, rx) = oneshot::channel::>>(); + // Shared between the success and error callbacks (only one ever fires). + let sender = Arc::new(Mutex::new(Some(tx))); + + Python::attach(|py| -> PyResult<()> { + // Create some deferred success/error callback functions that we will use to get + // the result from Python to Rust. + let success_sender = Arc::clone(&sender); + let on_success = PyCFunction::new_closure( + py, + None, + None, + move |args, _kwargs| -> PyResult> { + let value = args.get_item(0)?.unbind(); + if let Some(tx) = success_sender.lock().unwrap().take() { + let _ = tx.send(Ok(value)); + } + Ok(args.py().None()) + }, + )? + .unbind(); + + let error_sender = Arc::clone(&sender); + let on_error = PyCFunction::new_closure( + py, + None, + None, + move |args, _kwargs| -> PyResult> { + let err = failure_to_pyerr(&args.get_item(0)?); + if let Some(tx) = error_sender.lock().unwrap().take() { + let _ = tx.send(Err(err)); + } + Ok(args.py().None()) + }, + )? + .unbind(); + + // Wrap `make_awaitable` as a Python callable so we can hand it to + // `run_in_background`, which calls it (in the active logcontext) to produce + // the awaitable it then drives. + let awaitable_factory = PyCFunction::new_closure( + py, + None, + None, + move |args, _kwargs| -> PyResult> { + let py = args.py(); + Ok(make_awaitable(py)?.unbind()) + }, + )? + .unbind(); + + // Create a function that we will run with the Twisted reactor that will drive + // the Python awaitable. + let starter = PyCFunction::new_closure( + py, + None, + None, + move |args, _kwargs| -> PyResult> { + let py = args.py(); + + // We fire-and-forget using `run_in_background`. Re-using + // `run_in_background` also makes sure the awaitable gets run with the + // current logcontext while following the logcontext rules. + // + // FIXME: Currently runs in the sentinel logcontext because we don't manage it here + let deferred = logging_context_module(py)?.call_method1( + intern!(py, "run_in_background"), + (awaitable_factory.bind(py),), + ); + + let deferred = deferred?; + deferred.call_method1( + intern!(py, "addCallbacks"), + (on_success.bind(py), on_error.bind(py)), + )?; + Ok(py.None()) + }, + )?; + + reactor + .bind(py) + .call_method1(intern!(py, "callFromThread"), (starter,))?; + + Ok(()) + })?; + + match rx.await { + Ok(result) => result, + Err(_) => Err(PyRuntimeError::new_err( + "run_python_awaitable channel closed before the awaitable completed", + )), + } +} + +/// Convert a Twisted `Failure` (as passed to an Deferred errback) into a [`PyErr`]. +/// +/// A Twisted `Failure` carries the original exception instance in its `.value` +/// attribute, which we re-raise so callers see the real error. If the `Failure` is +/// mangled, we fallback to raising a generic [`PyRuntimeError`] explaining what we saw +/// instead. +fn failure_to_pyerr(failure: &Bound<'_, PyAny>) -> PyErr { + match failure.getattr(intern!(failure.py(), "value")) { + Ok(value) => PyErr::from_value(value), + Err(_) => PyRuntimeError::new_err(format!( + "Expected Python object passed here to be a Twisted `Failure` with a `value` attribute \ + but saw something else: {}", + failure + .str() + .map(|s| s.to_string_lossy().into_owned()) + .unwrap_or_else(|_| "".to_owned()), + )), + } +} + +static MAKE_DEFERRED_YIELDABLE: OnceCell> = OnceCell::new(); /// Given a deferred, make it follow the Synapse logcontext rules fn make_deferred_yieldable<'py>( py: Python<'py>, deferred: &Bound<'py, PyAny>, ) -> PyResult> { - let make_deferred_yieldable = MAKE_DEFERRED_YIELDABLE.get_or_init(|| { - let sys = PyModule::import(py, "synapse.logging.context").unwrap(); - let func = sys.getattr("make_deferred_yieldable").unwrap().unbind(); - func - }); + let make_deferred_yieldable = MAKE_DEFERRED_YIELDABLE.get_or_try_init(|| { + logging_context_module(py)? + .getattr("make_deferred_yieldable") + .map(Into::into) + })?; make_deferred_yieldable .call1(py, (deferred,))? @@ -133,6 +279,8 @@ fn make_deferred_yieldable<'py>( pub fn register_module(py: Python<'_>, _m: &Bound<'_, PyModule>) -> PyResult<()> { // Make sure we fail early if we can't load some modules defer(py)?; + // We can't check this here because of circular import issues + // logging_context_module(py)?; Ok(()) } From cee06f9620faf02fb2170c10f10bcdf034b1baa8 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 24 Jun 2026 18:34:51 -0500 Subject: [PATCH 03/32] Add Python implementation of the `DatabasePool` --- rust/src/storage/db/mod.rs | 2 + rust/src/storage/db/python_db_pool.rs | 343 ++++++++++++++++++++++++++ 2 files changed, 345 insertions(+) create mode 100644 rust/src/storage/db/python_db_pool.rs diff --git a/rust/src/storage/db/mod.rs b/rust/src/storage/db/mod.rs index a853eca6309..126faf15eab 100644 --- a/rust/src/storage/db/mod.rs +++ b/rust/src/storage/db/mod.rs @@ -19,6 +19,8 @@ use std::future::Future; use futures::future::BoxFuture; use futures::FutureExt; +pub mod python_db_pool; + /// A type-erased `run_interaction` callback. /// /// This is the dyn-compatible form of the `func` passed to diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs new file mode 100644 index 00000000000..28d23e57800 --- /dev/null +++ b/rust/src/storage/db/python_db_pool.rs @@ -0,0 +1,343 @@ +/* + * 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: + * . + * + */ + +//! A database pool that calls into Python to re-use the same database pool used on the +//! Python side. This is desirable because we want to avoid having two separate database +//! pools (one for Rust, one for Python) to avoid database connection exhaustion +//! problems. This is a stepping stone until all of our database interactions are in +//! Rust. +//! +//! We have these main classes: +//! - Database pool [`PythonDatabasePoolWrapper`] (implements [`DatabasePool`]) which +//! allows you to start a... +//! - transaction [`LoggingTransactionWrapper`] (implements [`Transaction`]) and query +//! the database. + +use std::future::Future; +use std::pin::pin; +use std::sync::{Arc, Mutex}; +use std::task::Poll; + +use anyhow::Context; +use pyo3::{ + exceptions::{PyRuntimeError, PyTypeError}, + intern, + prelude::*, + types::{PyBool, PyCFunction, PyFloat, PyInt, PyList, PyString}, +}; + +use crate::deferred::run_python_awaitable; +use crate::storage::db::{ + DatabasePool, DbRow, DbValue, ErasedInteraction, ErasedResult, Transaction, +}; + +/// The database engines we support in the Python side of Synapse +#[derive(Copy, Clone, Debug)] +pub enum DatabaseEngine { + Sqlite, + Postgres, +} + +impl DatabaseEngine { + //[inline] + pub fn supports_using_any_list(&self) -> bool { + match self { + DatabaseEngine::Sqlite => false, + DatabaseEngine::Postgres => true, + } + } +} + +/// Wrapper for a `DatabasePool` from the Python side of Synapse. +pub struct PythonDatabasePoolWrapper { + /// 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 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 { + async fn run_interaction_erased( + &self, + name: &'static str, + func: ErasedInteraction, + ) -> ErasedResult { + // `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 result back out through Python (remember, `func` returns an + // `ErasedResult`, not a `PyAny`), the callback stashes the result here and we + // pick it up once the deferred fires. + // + // Note the callback may run more than once (`runInteraction` retries on + // serialization/deadlock errors), so we only trust this slot once the + // deferred has fired, i.e. once the transaction has finally committed or + // failed. + let result_slot: Arc>> = Arc::new(Mutex::new(None)); + + // Build the callback that Python's `runInteraction` invokes on a DB + // thread with a `LoggingTransaction`, plus owned handles we can move onto + // the reactor thread. We drive `func` to completion in the callback; 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, database_pool_py, reactor) = Python::attach(|py| -> PyResult<_> { + 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::()?; + + // Since we expect people to only call `.await` on + // [`Transaction`] related methods (mentioned in the + // [`Transaction`] docstring) AND because there is no actual + // async work to suspend on in the Python [`Transaction`] + // (resolves synchonously), we can get away with polling once as + // it should immediately resolve to [`Poll::Ready`]. Getting + // [`Poll::Pending`] would be considered a programming error. + // + // Alternatively, we could just use `futures::executor::block_on` + // which is probably cleaner but a single-shot poll is more + // enforcing of the concept we want to represent. + match poll_once(func(&mut txn)) { + Poll::Ready(Ok(value)) => { + *callback_slot.lock().unwrap() = Some(Ok(value)); + Ok(py.None()) + } + Poll::Ready(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) + } + Poll::Pending => unreachable!( + "The `run_interaction` transaction callback future returned `Poll::Pending`, \ + but we expect Synapse Python database work to resolve synchronously. \ + This is a Synapse programming error: genuine async work is \ + not supported here.", + ), + } + }, + )? + .unbind(); + + Ok(( + callback, + self.database_pool_py.clone_ref(py), + self.reactor.clone_ref(py), + )) + }) + .map_err(anyhow::Error::from)?; + + // Use `runInteraction` directly + let run_interaction_outcome = run_python_awaitable(reactor, move |py| { + database_pool_py + .bind(py) + .call_method1(intern!(py, "runInteraction"), (name, callback.bind(py))) + }) + .await; + + // Return the result we captured based on if `runInteraction` was successful + let captured_result = result_slot.lock().unwrap().take(); + match run_interaction_outcome { + // Only return the `captured_result` if `runInteraction` succeeded. We don't + // want to accidentally return a successful result when the transaction + // actually failed to commit. + Ok(_) => match captured_result { + Some(result) => result, + // This is unexpected as we either expect `runInteraction` to have + // completed successfully and run the provided `callback` which runs the + // `func` and we capture a result or it fails. + None => Err(anyhow::anyhow!( + "Expected to capture result after running `runInteraction` and seeing it succeed (but saw nothing). \ + This is a Synapse programming error." + )), + }, + Err(py_err) => Err(anyhow::Error::from(py_err)).with_context(|| format!("run_interaction(name={}) failed", name)), + } + } +} + +/// Poll a future exactly once. +/// +/// Returns [`Poll::Ready`] if the future resolves immediately, or [`Poll::Pending`] if +/// it would need to suspend. We use this where a future is expected to never genuinely +/// suspend and want to enforce/check that, rather than driving it to completion with a +/// blocking executor like `futures::executor::block_on`. +fn poll_once(future: F) -> Poll { + let mut future = pin!(future); + let waker = futures::task::noop_waker(); + let mut cx = std::task::Context::from_waker(&waker); + future.as_mut().poll(&mut cx) +} + +/// 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:#}")) +} + +/// Given a Python `LoggingTransaction`, figures out the database engine that backs it +fn detect_engine(txn_py: &Bound<'_, PyAny>) -> PyResult { + let name = txn_py + .getattr("database_engine") + .expect("`LoggingTransaction` must have `database_engine` attr") + .get_type() + .name() + .expect("`database_engine` type must have a name") + .to_str() + .expect("`database_engine` type name must be valid UTF-8") + .to_owned(); + + Ok(match name.as_str() { + "PostgresEngine" => DatabaseEngine::Postgres, + "Sqlite3Engine" => DatabaseEngine::Sqlite, + other => unimplemented!( + "Unknown database engine {other:?}. This is a Synapse programming error." + ), + }) +} + +/// Wrapper for a `LoggingTransaction` from the Python side of Synapse. +pub struct LoggingTransactionWrapper { + /// The underlying `LoggingTransaction` + /// + /// We purposely avoid `Bound<'py, PyAny>` so it can be stored and moved freely + /// across threads (like to extract it from the `new_transaction(...)` callback). + /// You will need to acquire your own `py` and bind it using + /// `logging_transaction_py.bind(py)` to do anything useful. + logging_transaction_py: Py, + + /// Disambiguate which underlying database engine we're working with + /// + /// Some features are only available on Postgres vs SQLite and the queries need to + /// be differentiated (for compatibility or performance reasons). + pub database_engine: DatabaseEngine, +} + +impl<'a, 'py> FromPyObject<'a, 'py> for LoggingTransactionWrapper { + type Error = PyErr; + + /// Extract from a Python `LoggingTransaction` passed as an argument. + fn extract(logging_transaction_py: Borrowed<'a, 'py, PyAny>) -> PyResult { + let database_engine = detect_engine(&logging_transaction_py.to_owned())?; + Ok(Self { + logging_transaction_py: logging_transaction_py.to_owned().unbind(), + database_engine, + }) + } +} + +impl LoggingTransactionWrapper { + fn execute<'py>( + &mut self, + py: Python<'py>, + sql: &str, + args: &Bound<'py, PyAny>, + ) -> PyResult<()> { + let execute_fn = self + .logging_transaction_py + .bind(py) + .getattr(intern!(py, "execute"))?; + execute_fn.call1((sql, args))?; + Ok(()) + } +} + +#[async_trait::async_trait] +impl Transaction for LoggingTransactionWrapper { + 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`. + // + // We don't need to do anything to the SQL as the `?`-style arg placeholders + // already align with what `LoggingTransaction` expects. + let args = PyList::new(py, args)?; + self.execute(py, sql, args.as_any())?; + + // Pull the rows back out, converting each cell from its Python type + // into the engine-agnostic `DbValue` representation as we go. + 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: DbRow = Vec::new(); + for cell in row_py.try_iter()? { + row.push(py_cell_to_value(&cell?)?); + } + rows.push(row); + } + + Ok(rows) + }) + .map_err(anyhow::Error::from) + } +} + +/// Convert a single cell from a Python row into a backend-agnostic [`DbValue`] by +/// inspecting its Python type (the pyo3 equivalent of `isinstance` checks). +fn py_cell_to_value(cell: &Bound<'_, PyAny>) -> PyResult { + // `None` maps to SQL `NULL`. + if cell.is_none() { + return Ok(DbValue::Null); + } + + // A `bool` *is* an `int` in SQLite, so ensure we try `bool` first. + if let Ok(b) = cell.cast::() { + Ok(DbValue::Bool(b.extract()?)) + } else if let Ok(i) = cell.cast::() { + Ok(DbValue::Int(i.extract()?)) + } else if let Ok(f) = cell.cast::() { + Ok(DbValue::Float(f.extract()?)) + } else if let Ok(s) = cell.cast::() { + Ok(DbValue::Text(s.to_string())) + } else { + Err(PyTypeError::new_err(format!( + "unsupported column type {} returned from the database", + cell.get_type().name()? + ))) + } +} From 9225c8e31492df4416b8f90e24309f19487cb4bd Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 24 Jun 2026 18:36:58 -0500 Subject: [PATCH 04/32] Add `Store` with example usage of `run_interaction(...)` --- rust/src/storage/mod.rs | 1 + rust/src/storage/store.rs | 106 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 107 insertions(+) create mode 100644 rust/src/storage/store.rs diff --git a/rust/src/storage/mod.rs b/rust/src/storage/mod.rs index ac2c181452f..735fcecb266 100644 --- a/rust/src/storage/mod.rs +++ b/rust/src/storage/mod.rs @@ -14,3 +14,4 @@ */ pub mod db; +pub mod store; diff --git a/rust/src/storage/store.rs b/rust/src/storage/store.rs new file mode 100644 index 00000000000..e67b25e496c --- /dev/null +++ b/rust/src/storage/store.rs @@ -0,0 +1,106 @@ +/* + * 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: + * . + * + */ + +use std::sync::Arc; + +use futures::FutureExt; +use serde::Serialize; + +use crate::storage::db::{DatabasePool, DatabasePoolExt, DbRowExt}; + +/// Currently supported per-user features +#[derive(Serialize, Debug)] +pub enum PerUserExperimentalFeature { + #[serde(rename = "msc3881")] + MSC3881, + #[serde(rename = "msc3575")] + MSC3575, + #[serde(rename = "msc4222")] + MSC4222, +} + +impl std::fmt::Display for PerUserExperimentalFeature { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + write!( + f, + "{}", + // Serialize so we can use the serde name of the variant as the source of truth + serde_json::to_string(self) + .unwrap_or_else(|err| format!( + "", + self, err + )) + // Remove the surrounding quotes from JSON serialization + .trim_matches('"') + ) + } +} + +pub struct Store { + pub db_pool: Box, +} + +impl Store { + /// Checks whether a given feature is enabled/disabled for this user + /// + /// If there is no entry, returns None + pub async fn is_feature_enabled_for_user( + &self, + user_id: &str, + feature: PerUserExperimentalFeature, + ) -> Result, anyhow::Error> { + // It's not enabled globally, so check whether it's enabled per-user. + // + // We need owned copies to move into the callback because it is `'static` (it + // may be moved to another thread). We use `Arc` rather than `String` so + // the per-call clone is just a cheap refcount bump rather than a fresh + // allocation. + let user_id: Arc = user_id.into(); + let feature: Arc = feature.to_string().into(); + + 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_ref(), feature.as_ref()], + ) + .await?; + + let enabled = match &rows[..] { + // No row for this user + [] => None, + // Otherwise, we should only find a single row for this (user, feature) + [row] => Some(row.try_get(0)?), + _ => { + panic!("Programming error (SQL query probably doens't match our expectations)") + } + }; + + Ok(enabled) + } + .boxed() + }) + .await?; + + Ok(is_feature_enabled_for_user) + } +} From d978a66190f972746d3db1dfb1efc41c16c9432f Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 24 Jun 2026 18:43:30 -0500 Subject: [PATCH 05/32] Update `/versions` endpoint to use Rust `/versions` is a good candidate as its is the simplest endpoint that uses the database. We use `tests/rest/client/test_versions.py` as a sanity check that all of this new Rust database access works. --- rust/src/config/mod.rs | 103 +++++++++ rust/src/handlers/mod.rs | 95 +++++++++ rust/src/handlers/versions.rs | 328 +++++++++++++++++++++++++++++ rust/src/lib.rs | 3 + synapse/config/room.py | 10 +- synapse/rest/client/versions.py | 137 +----------- synapse/server.py | 5 + synapse/synapse_rust/handlers.pyi | 35 +++ tests/rest/client/test_versions.py | 179 ++++++++++++++++ tests/server.py | 51 ++++- 10 files changed, 806 insertions(+), 140 deletions(-) create mode 100644 rust/src/config/mod.rs create mode 100644 rust/src/handlers/mod.rs create mode 100644 rust/src/handlers/versions.rs create mode 100644 synapse/synapse_rust/handlers.pyi create mode 100644 tests/rest/client/test_versions.py diff --git a/rust/src/config/mod.rs b/rust/src/config/mod.rs new file mode 100644 index 00000000000..709b4428952 --- /dev/null +++ b/rust/src/config/mod.rs @@ -0,0 +1,103 @@ +/* + * 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: + * . + * + */ + +use std::collections::BTreeSet; +use std::str::FromStr; + +use pyo3::{exceptions::PyRuntimeError, prelude::*}; + +/// A Rust-side view of Synapse's Python `HomeServerConfig`. +/// +/// This only mirrors the subset of config that the Rust handlers need, rather +/// than the whole thing. Thanks to `#[derive(FromPyObject)]`, each field is +/// pulled directly off the corresponding attribute of the Python `config` +/// object, so you can populate it in one shot with +/// `homeserver.getattr("config")?.extract()?`. +#[derive(FromPyObject, Clone)] +pub struct SynapseHomeServerConfig { + pub room: RoomConfig, + pub auth: AuthConfig, + pub server: ServerConfig, + pub experimental: ExperimentalConfig, +} + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub enum RoomCreationPreset { + PrivateChat, + PublicChat, + TrustedPrivateChat, +} + +impl FromStr for RoomCreationPreset { + type Err = PyErr; + + fn from_str(s: &str) -> Result { + Ok(match s { + "private_chat" => RoomCreationPreset::PrivateChat, + "public_chat" => RoomCreationPreset::PublicChat, + "trusted_private_chat" => RoomCreationPreset::TrustedPrivateChat, + other => { + return Err(PyRuntimeError::new_err(format!( + "Unknown variant {other:?} does not translate to `RoomCreationPreset`. \ + This is a Synapse programming error." + ))) + } + }) + } +} + +impl<'a, 'py> FromPyObject<'a, 'py> for RoomCreationPreset { + type Error = PyErr; + + fn extract(value: Borrowed<'a, 'py, PyAny>) -> PyResult { + value.extract::<&str>()?.parse() + } +} + +#[derive(FromPyObject, Clone)] +pub struct RoomConfig { + pub encryption_enabled_by_default_for_room_presets: BTreeSet, +} + +#[derive(FromPyObject, Clone)] +pub struct AuthConfig { + pub login_via_existing_enabled: bool, +} +#[derive(FromPyObject, Clone)] +pub struct ServerConfig { + pub max_event_delay_ms: Option, +} + +#[derive(FromPyObject, Clone)] +pub struct ExperimentalConfig { + pub msc3026_enabled: bool, + pub msc3773_enabled: bool, + pub msc2815_enabled: bool, + pub msc3881_enabled: bool, + pub msc3874_enabled: bool, + pub msc3912_enabled: bool, + pub msc3391_enabled: bool, + pub msc4069_profile_inhibit_propagation: bool, + pub msc4028_push_encrypted_events: bool, + pub msc4108_enabled: bool, + pub msc4108_delegation_endpoint: Option, + pub msc3575_enabled: bool, + pub msc4133_enabled: bool, + pub msc4155_enabled: bool, + pub msc4306_enabled: bool, + pub msc4169_enabled: bool, + pub msc4354_enabled: bool, + pub msc4222_enabled: bool, +} diff --git a/rust/src/handlers/mod.rs b/rust/src/handlers/mod.rs new file mode 100644 index 00000000000..d5604231669 --- /dev/null +++ b/rust/src/handlers/mod.rs @@ -0,0 +1,95 @@ +/* + * 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: + * . + * + */ + +use std::sync::Arc; + +use pyo3::{ + prelude::*, + types::{PyAnyMethods, PyModule, PyModuleMethods}, + Bound, PyResult, Python, +}; + +use crate::config::SynapseHomeServerConfig; +use crate::storage::db::python_db_pool::PythonDatabasePoolWrapper; +use crate::storage::store::Store; + +pub mod versions; + +#[pyclass] +struct RustHandlers { + versions: Py, +} + +#[pymethods] +impl RustHandlers { + #[new] + #[pyo3(signature = (homeserver))] + pub fn py_new(py: Python<'_>, homeserver: &Bound<'_, PyAny>) -> PyResult { + let config: SynapseHomeServerConfig = 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_py: Py = homeserver + .call_method0("get_datastores")? + .getattr("main")? + .getattr("db_pool")? + .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 { + db_pool: Box::new(db_pool), + }); + + let global_unstable_feature_map = Arc::new( + versions::synapse_config_to_global_unstable_feature_map(&config), + ); + + let versions = Py::new( + py, + versions::VersionsHandler { + global_unstable_feature_map: Arc::clone(&global_unstable_feature_map), + store: Arc::clone(&store), + reactor: reactor.clone_ref(py), + }, + )?; + + Ok(RustHandlers { versions }) + } + + #[getter] + fn versions(&self, py: Python<'_>) -> Py { + self.versions.clone_ref(py) + } +} + +/// Called when registering modules with python. +pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { + let child_module = PyModule::new(py, "handlers")?; + child_module.add_class::()?; + + m.add_submodule(&child_module)?; + + // We need to manually add the module to sys.modules to make `from + // synapse.synapse_rust import push` work. + py.import("sys")? + .getattr("modules")? + .set_item("synapse.synapse_rust.handlers", child_module)?; + + Ok(()) +} diff --git a/rust/src/handlers/versions.rs b/rust/src/handlers/versions.rs new file mode 100644 index 00000000000..a723735fc66 --- /dev/null +++ b/rust/src/handlers/versions.rs @@ -0,0 +1,328 @@ +/* + * 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: + * . + * + */ + +use std::sync::Arc; + +use pyo3::prelude::*; +use pythonize::{pythonize, PythonizeError}; +use serde::{Deserialize, Serialize}; + +use crate::config::{RoomCreationPreset, SynapseHomeServerConfig}; +use crate::deferred::create_deferred; +use crate::storage::store::{PerUserExperimentalFeature, Store}; + +/// `GET /_matrix/client/versions` response +#[derive(Serialize, Deserialize, Clone, Debug)] +struct VersionsResponse { + versions: Vec, + /// as per MSC1497 + 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 global_unstable_feature_map: Arc, + 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, 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 global_unstable_feature_map = Arc::clone(&self.global_unstable_feature_map); + + create_deferred(py, self.reactor.bind(py), async move { + build_versions_response(&store, &global_unstable_feature_map, 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. +/// +/// Args: +/// * store +/// * global_unstable_feature_map: The global values before any per-user overrides +/// * user_id: The user making the request +async fn build_versions_response( + store: &Store, + global_unstable_feature_map: &UnstableFeatureMap, + user_id: Option<&str>, +) -> Result { + let msc3881_enabled = match user_id { + Some(user_id) => { + // Don't both looking anything up if it's enabled for everyone + if global_unstable_feature_map.msc3881 { + true + } else { + // Look up whether it's explicitly enabled/disabled for this user + store + .is_feature_enabled_for_user(user_id, PerUserExperimentalFeature::MSC3881) + .await? + // Default to false if there is no entry for this user + .unwrap_or(false) + } + } + None => global_unstable_feature_map.msc3881, + }; + + let msc3575_enabled = match user_id { + Some(user_id) => { + // Don't both looking anything up if it's enabled for everyone + if global_unstable_feature_map.msc3575 { + true + } else { + // Look up whether it's explicitly enabled/disabled for this user + store + .is_feature_enabled_for_user(user_id, PerUserExperimentalFeature::MSC3575) + .await? + // Default to false if there is no entry for this user + .unwrap_or(false) + } + } + None => global_unstable_feature_map.msc3575, + }; + + let unstable_feature_map = UnstableFeatureMap { + msc3575: msc3575_enabled, + msc3881: msc3881_enabled, + // The clone here isn't the best but better than manually composing things + ..global_unstable_feature_map.clone() + }; + + 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 + // backwards compatible with r0.2.0. But need to check how + // conscientious we've been in compatibility, and decide whether the + // middle number is the major revision when at 0.X.Y (as opposed to + // X.Y.Z). And we need to decide whether it's fair to make clients + // parse the version string to figure out what's going on. + "r0.0.1".to_string(), + "r0.1.0".to_string(), + "r0.2.0".to_string(), + "r0.3.0".to_string(), + "r0.4.0".to_string(), + "r0.5.0".to_string(), + "r0.6.0".to_string(), + "r0.6.1".to_string(), + "v1.1".to_string(), + "v1.2".to_string(), + "v1.3".to_string(), + "v1.4".to_string(), + "v1.5".to_string(), + "v1.6".to_string(), + "v1.7".to_string(), + "v1.8".to_string(), + "v1.9".to_string(), + "v1.10".to_string(), + "v1.11".to_string(), + "v1.12".to_string(), + ]), + unstable_features: serde_json::from_value(serde_json::to_value(unstable_feature_map)?)?, + }) +} + +/// Experimental features the server supports +#[derive(Serialize, Debug, Clone)] +pub struct UnstableFeatureMap { + /// Implements support for label-based filtering as described in + /// MSC2326. + #[serde(rename = "org.matrix.label_based_filtering")] + msc2326: bool, + /// Implements support for cross signing as described in MSC1756 + #[serde(rename = "org.matrix.e2e_cross_signing")] + msc1756: bool, + /// Implements additional endpoints as described in MSC2432 + #[serde(rename = "org.matrix.msc2432")] + msc2432: bool, + /// Implements additional endpoints as described in MSC2666 + #[serde(rename = "uk.half-shot.msc2666.query_mutual_rooms.stable")] + msc2666: bool, + // Supports the busy presence state described in MSC3026. + #[serde(rename = "org.matrix.msc3026.busy_presence")] + msc3026: bool, + /// Supports receiving private read receipts as per MSC2285 + // TODO: Remove when MSC2285 becomes a part of the spec + #[serde(rename = "org.matrix.msc2285.stable")] + msc2285: bool, + /// Supports filtering of /publicRooms by room type as per MSC3827 + #[serde(rename = "org.matrix.msc3827.stable")] + msc3827: bool, + /// Adds support for thread relations, per MSC3440. + // TODO: remove when "v1.3" is added above + #[serde(rename = "org.matrix.msc3440.stable")] + msc3440: bool, + /// Support for thread read receipts & notification counts. + #[serde(rename = "org.matrix.msc3771")] + msc3771: bool, + #[serde(rename = "org.matrix.msc3773")] + msc3773: bool, + /// Allows moderators to fetch redacted event content as described in MSC2815 + #[serde(rename = "fi.mau.msc2815")] + msc2815: bool, + /// Adds a ping endpoint for appservices to check HS->AS connection + // TODO: remove when "v1.7" is added above + #[serde(rename = "fi.mau.msc2659.stable")] + msc2659: bool, + // TODO: this is no longer needed once unstable MSC3882 does not need to be supported: + #[serde(rename = "org.matrix.msc3882")] + msc3882: bool, + /// Adds support for remotely enabling/disabling pushers, as per MSC3881 + #[serde(rename = "org.matrix.msc3881")] + msc3881: bool, + /// Adds support for filtering /messages by event relation. + #[serde(rename = "org.matrix.msc3874")] + msc3874: bool, + // Adds support for relation-based redactions as per MSC3912. + #[serde(rename = "org.matrix.msc3912")] + msc3912: bool, + /// Whether recursively provide relations is supported. + // TODO This is no longer needed once unstable MSC3981 does not need to be supported. + #[serde(rename = "org.matrix.msc3981")] + msc3981: bool, + /// Adds support for deleting account data. + #[serde(rename = "org.matrix.msc3391")] + msc3391: bool, + /// Allows clients to inhibit profile update propagation. + #[serde(rename = "org.matrix.msc4069")] + msc4069: bool, + // Allows clients to handle push for encrypted events. + #[serde(rename = "org.matrix.msc4028")] + msc4028: bool, + /// MSC4108: Mechanism to allow OIDC sign in and E2EE set up via QR code - 2024 version + #[serde(rename = "org.matrix.msc4108")] + msc4108: bool, + /// MSC4140: Delayed events + #[serde(rename = "org.matrix.msc4140")] + msc4140: bool, + /// Simplified sliding sync + #[serde(rename = "org.matrix.simplified_msc3575")] + msc3575: bool, + /// Arbitrary key-value profile fields. + #[serde(rename = "uk.tcpip.msc4133")] + msc4133: bool, + /// Arbitrary key-value profile fields (stable identifier) + #[serde(rename = "uk.tcpip.msc4133.stable")] + msc4133_stable: bool, + /// MSC4155: Invite filtering + #[serde(rename = "org.matrix.msc4155")] + msc4155: bool, + /// MSC4306: Support for thread subscriptions + #[serde(rename = "org.matrix.msc4306")] + msc4306: bool, + /// MSC4169: Backwards-compatible redaction sending using `/send` + #[serde(rename = "com.beeper.msc4169")] + msc4169: bool, + /// MSC4354: Sticky events + #[serde(rename = "org.matrix.msc4354")] + msc4354: bool, + /// MSC4380: Invite blocking + #[serde(rename = "org.matrix.msc4380.stable")] + msc4380: bool, + // MSC4445: Sync timeline order + #[serde(rename = "org.matrix.msc4445.initial_sync_timeline_topological_ordering")] + msc4445_initial_sync_timeline_topological_ordering: bool, + + // Whether new rooms will be set to encrypted or not (based on presets). + #[serde(rename = "io.element.e2ee_forced.public")] + e2ee_forced_public: bool, + #[serde(rename = "io.element.e2ee_forced.private")] + e2ee_forced_private: bool, + #[serde(rename = "io.element.e2ee_forced.trusted_private")] + e2ee_forced_trusted_private: bool, +} + +/// Convert from [`SynapseHomeServerConfig`] to the global defaults for unstable features that the +/// server supports [`UnstableFeatureMap`] +pub fn synapse_config_to_global_unstable_feature_map( + config: &SynapseHomeServerConfig, +) -> UnstableFeatureMap { + UnstableFeatureMap { + msc2326: true, + msc1756: true, + msc2432: true, + msc2666: true, + msc3026: config.experimental.msc3026_enabled, + msc2285: true, + msc3827: true, + msc3440: true, + msc3771: true, + msc3773: config.experimental.msc3773_enabled, + msc2815: config.experimental.msc2815_enabled, + msc2659: true, + msc3882: config.auth.login_via_existing_enabled, + msc3881: config.experimental.msc3881_enabled, + msc3874: config.experimental.msc3874_enabled, + msc3912: config.experimental.msc3912_enabled, + msc3981: true, + msc3391: config.experimental.msc3391_enabled, + msc4069: config.experimental.msc4069_profile_inhibit_propagation, + msc4028: config.experimental.msc4028_push_encrypted_events, + msc4108: config.experimental.msc4108_enabled + || (config.experimental.msc4108_delegation_endpoint.is_some()), + msc4140: config + .server + .max_event_delay_ms + .is_some_and(|max_event_delay_ms| max_event_delay_ms > 0), + msc3575: config.experimental.msc3575_enabled, + msc4133: config.experimental.msc4133_enabled, + msc4133_stable: true, + msc4155: config.experimental.msc4155_enabled, + msc4306: config.experimental.msc4306_enabled, + msc4169: config.experimental.msc4169_enabled, + msc4354: config.experimental.msc4354_enabled, + msc4380: true, + msc4445_initial_sync_timeline_topological_ordering: true, + e2ee_forced_public: config + .room + .encryption_enabled_by_default_for_room_presets + .contains(&RoomCreationPreset::PublicChat), + e2ee_forced_private: config + .room + .encryption_enabled_by_default_for_room_presets + .contains(&RoomCreationPreset::PrivateChat), + e2ee_forced_trusted_private: config + .room + .encryption_enabled_by_default_for_room_presets + .contains(&RoomCreationPreset::TrustedPrivateChat), + } +} diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 6e3ee06f75d..28783afbbac 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -6,10 +6,12 @@ use pyo3_log::ResetHandle; pub mod acl; pub mod canonical_json; +pub mod config; pub mod deferred; pub mod duration; pub mod errors; pub mod events; +pub mod handlers; pub mod http; pub mod http_client; pub mod identifier; @@ -71,6 +73,7 @@ fn synapse_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { deferred::register_module(py, m)?; push::register_module(py, m)?; events::register_module(py, m)?; + handlers::register_module(py, m)?; http_client::register_module(py, m)?; rendezvous::register_module(py, m)?; msc4388_rendezvous::register_module(py, m)?; diff --git a/synapse/config/room.py b/synapse/config/room.py index e698c7bafd6..6a1f4d1eb8b 100644 --- a/synapse/config/room.py +++ b/synapse/config/room.py @@ -48,23 +48,23 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None: RoomDefaultEncryptionTypes.OFF, ) if encryption_for_room_type == RoomDefaultEncryptionTypes.ALL: - self.encryption_enabled_by_default_for_room_presets = [ + self.encryption_enabled_by_default_for_room_presets = { RoomCreationPreset.PRIVATE_CHAT, RoomCreationPreset.TRUSTED_PRIVATE_CHAT, RoomCreationPreset.PUBLIC_CHAT, - ] + } elif encryption_for_room_type == RoomDefaultEncryptionTypes.INVITE: - self.encryption_enabled_by_default_for_room_presets = [ + self.encryption_enabled_by_default_for_room_presets = { RoomCreationPreset.PRIVATE_CHAT, RoomCreationPreset.TRUSTED_PRIVATE_CHAT, - ] + } elif ( encryption_for_room_type == RoomDefaultEncryptionTypes.OFF or encryption_for_room_type is False ): # PyYAML translates "off" into False if it's unquoted, so we also need to # check for encryption_for_room_type being False. - self.encryption_enabled_by_default_for_room_presets = [] + self.encryption_enabled_by_default_for_room_presets = set() else: raise ConfigError( "Invalid value for encryption_enabled_by_default_for_room_type" diff --git a/synapse/rest/client/versions.py b/synapse/rest/client/versions.py index bb1711f2cf2..9312a616dab 100644 --- a/synapse/rest/client/versions.py +++ b/synapse/rest/client/versions.py @@ -25,11 +25,9 @@ import re from typing import TYPE_CHECKING -from synapse.api.constants import RoomCreationPreset from synapse.http.server import HttpServer from synapse.http.servlet import RestServlet from synapse.http.site import SynapseRequest -from synapse.rest.admin.experimental_features import ExperimentalFeature from synapse.types import JsonDict if TYPE_CHECKING: @@ -47,25 +45,10 @@ def __init__(self, hs: "HomeServer"): self.config = hs.config self.auth = hs.get_auth() self.store = hs.get_datastores().main - - # Calculate these once since they shouldn't change after start-up. - self.e2ee_forced_public = ( - RoomCreationPreset.PUBLIC_CHAT - in self.config.room.encryption_enabled_by_default_for_room_presets - ) - self.e2ee_forced_private = ( - RoomCreationPreset.PRIVATE_CHAT - in self.config.room.encryption_enabled_by_default_for_room_presets - ) - self.e2ee_forced_trusted_private = ( - RoomCreationPreset.TRUSTED_PRIVATE_CHAT - in self.config.room.encryption_enabled_by_default_for_room_presets - ) + self.rust_handlers = hs.get_rust_handlers() async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]: - msc3881_enabled = self.config.experimental.msc3881_enabled - msc3575_enabled = self.config.experimental.msc3575_enabled - + user_id = None if self.auth.has_access_token(request): requester = await self.auth.get_user_by_req( request, @@ -74,13 +57,6 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]: allow_expired=True, ) user_id = requester.user.to_string() - - msc3881_enabled = await self.store.is_feature_enabled( - user_id, ExperimentalFeature.MSC3881 - ) - msc3575_enabled = await self.store.is_feature_enabled( - user_id, ExperimentalFeature.MSC3575 - ) else: # Allow caching of unauthenticated responses, as they only depend # on server configuration which rarely changes. @@ -102,114 +78,11 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]: # 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) + return ( 200, - { - "versions": [ - # 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 - # backwards compatible with r0.2.0. But need to check how - # conscientious we've been in compatibility, and decide whether the - # middle number is the major revision when at 0.X.Y (as opposed to - # X.Y.Z). And we need to decide whether it's fair to make clients - # parse the version string to figure out what's going on. - "r0.0.1", - "r0.1.0", - "r0.2.0", - "r0.3.0", - "r0.4.0", - "r0.5.0", - "r0.6.0", - "r0.6.1", - "v1.1", - "v1.2", - "v1.3", - "v1.4", - "v1.5", - "v1.6", - "v1.7", - "v1.8", - "v1.9", - "v1.10", - "v1.11", - "v1.12", - ], - # as per MSC1497: - "unstable_features": { - # Implements support for label-based filtering as described in - # MSC2326. - "org.matrix.label_based_filtering": True, - # Implements support for cross signing as described in MSC1756 - "org.matrix.e2e_cross_signing": True, - # Implements additional endpoints as described in MSC2432 - "org.matrix.msc2432": True, - # Implements additional endpoints as described in MSC2666 - "uk.half-shot.msc2666.query_mutual_rooms.stable": True, - # Whether new rooms will be set to encrypted or not (based on presets). - "io.element.e2ee_forced.public": self.e2ee_forced_public, - "io.element.e2ee_forced.private": self.e2ee_forced_private, - "io.element.e2ee_forced.trusted_private": self.e2ee_forced_trusted_private, - # Supports the busy presence state described in MSC3026. - "org.matrix.msc3026.busy_presence": self.config.experimental.msc3026_enabled, - # Supports receiving private read receipts as per MSC2285 - "org.matrix.msc2285.stable": True, # TODO: Remove when MSC2285 becomes a part of the spec - # Supports filtering of /publicRooms by room type as per MSC3827 - "org.matrix.msc3827.stable": True, - # Adds support for thread relations, per MSC3440. - "org.matrix.msc3440.stable": True, # TODO: remove when "v1.3" is added above - # Support for thread read receipts & notification counts. - "org.matrix.msc3771": True, - "org.matrix.msc3773": self.config.experimental.msc3773_enabled, - # Allows moderators to fetch redacted event content as described in MSC2815 - "fi.mau.msc2815": self.config.experimental.msc2815_enabled, - # Adds a ping endpoint for appservices to check HS->AS connection - "fi.mau.msc2659.stable": True, # TODO: remove when "v1.7" is added above - # TODO: this is no longer needed once unstable MSC3882 does not need to be supported: - "org.matrix.msc3882": self.config.auth.login_via_existing_enabled, - # Adds support for remotely enabling/disabling pushers, as per MSC3881 - "org.matrix.msc3881": msc3881_enabled, - # Adds support for filtering /messages by event relation. - "org.matrix.msc3874": self.config.experimental.msc3874_enabled, - # Adds support for relation-based redactions as per MSC3912. - "org.matrix.msc3912": self.config.experimental.msc3912_enabled, - # Whether recursively provide relations is supported. - # TODO This is no longer needed once unstable MSC3981 does not need to be supported. - "org.matrix.msc3981": True, - # Adds support for deleting account data. - "org.matrix.msc3391": self.config.experimental.msc3391_enabled, - # Allows clients to inhibit profile update propagation. - "org.matrix.msc4069": self.config.experimental.msc4069_profile_inhibit_propagation, - # Allows clients to handle push for encrypted events. - "org.matrix.msc4028": self.config.experimental.msc4028_push_encrypted_events, - # MSC4108: Mechanism to allow OIDC sign in and E2EE set up via QR code - 2024 version - "org.matrix.msc4108": ( - self.config.experimental.msc4108_enabled - or ( - self.config.experimental.msc4108_delegation_endpoint - is not None - ) - ), - # MSC4140: Delayed events - "org.matrix.msc4140": bool(self.config.server.max_event_delay_ms), - # Simplified sliding sync - "org.matrix.simplified_msc3575": msc3575_enabled, - # Arbitrary key-value profile fields. - "uk.tcpip.msc4133": self.config.experimental.msc4133_enabled, - "uk.tcpip.msc4133.stable": True, - # MSC4155: Invite filtering - "org.matrix.msc4155": self.config.experimental.msc4155_enabled, - # MSC4306: Support for thread subscriptions - "org.matrix.msc4306": self.config.experimental.msc4306_enabled, - # MSC4169: Backwards-compatible redaction sending using `/send` - "com.beeper.msc4169": self.config.experimental.msc4169_enabled, - # MSC4354: Sticky events - "org.matrix.msc4354": self.config.experimental.msc4354_enabled, - # MSC4380: Invite blocking - "org.matrix.msc4380.stable": True, - # MSC4445: Sync timeline order - "org.matrix.msc4445.initial_sync_timeline_topological_ordering": True, - }, - }, + versions_response_body, ) diff --git a/synapse/server.py b/synapse/server.py index 8bf19f11b5d..635e444eb63 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -174,6 +174,7 @@ 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 @@ -963,6 +964,10 @@ def get_send_email_handler(self) -> SendEmailHandler: def get_set_password_handler(self) -> SetPasswordHandler: return SetPasswordHandler(self) + @cache_in_self + def get_rust_handlers(self) -> RustHandlers: + return RustHandlers(self) + @cache_in_self def get_event_sources(self) -> EventSources: return EventSources(self) diff --git a/synapse/synapse_rust/handlers.pyi b/synapse/synapse_rust/handlers.pyi new file mode 100644 index 00000000000..4c4f71b7bdb --- /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: ... diff --git a/tests/rest/client/test_versions.py b/tests/rest/client/test_versions.py new file mode 100644 index 00000000000..429250e58e5 --- /dev/null +++ b/tests/rest/client/test_versions.py @@ -0,0 +1,179 @@ +# 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: +# . + +import logging + +from twisted.internet.testing import MemoryReactor + +from synapse.rest import admin +from synapse.rest.client import login, versions +from synapse.server import HomeServer +from synapse.synapse_rust.http_client import HttpClient +from synapse.types import JsonDict +from synapse.util.clock import Clock + +from tests import unittest + +logger = logging.getLogger(__name__) + + +class VersionsTestCase(unittest.HomeserverTestCase): + """ + Test `VersionsRestServlet` + """ + + servlets = [ + admin.register_servlets, + login.register_servlets, + versions.register_servlets, + ] + + def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: + hs = self.setup_test_homeserver() + + # XXX: We must create the Rust HTTP client before we call `reactor.run()` below. + # Twisted's `MemoryReactor` doesn't invoke `callWhenRunning` callbacks if it's + # already running and we rely on that to start the Tokio thread pool in Rust. In + # the future, this may not matter, see https://github.com/twisted/twisted/pull/12514 + self._http_client = hs.get_proxied_http_client() + _ = HttpClient( + reactor=hs.get_reactor(), + user_agent=self._http_client.user_agent.decode("utf8"), + ) + + # This triggers the server startup hooks, which starts the Tokio thread pool + reactor.run() + + return hs + + def tearDown(self) -> None: + # MemoryReactor doesn't trigger the shutdown phases, and we want the + # Tokio thread pool to be stopped + # XXX: This logic should probably get moved somewhere else + shutdown_triggers = self.reactor.triggers.get("shutdown", {}) + for phase in ["before", "during", "after"]: + triggers = shutdown_triggers.get(phase, []) + for callbable, args, kwargs in triggers: + callbable(*args, **kwargs) + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.admin_user = self.register_user("admin", "pass", admin=True) + self.admin_user_tok = self.login("admin", "pass") + + def test_unauthenticated(self) -> None: + channel = self.make_request( + "GET", + "/_matrix/client/versions", + content={}, + ) + self.assertEqual(channel.code, 200, channel.result) + self._sanity_check_versions_response(channel.json_body) + + def test_authenticated(self) -> None: + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + + channel = self.make_request( + "GET", + "/_matrix/client/versions", + content={}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) + self._sanity_check_versions_response(channel.json_body) + + def test_authenticated_with_per_user_feature(self) -> None: + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + # Sanity check that the experimental feature should not be enabled yet + channel = self.make_request( + "GET", + "/_matrix/client/versions", + content={}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) + self._sanity_check_versions_response(channel.json_body) + self.assertEqual( + channel.json_body["unstable_features"]["org.matrix.msc3881"], + False, + channel.json_body, + ) + + # Enable the feature for this specific user + self._enable_experimental_feature_for_user( + target_user_id=user1_id, features={"msc3881": True} + ) + + # The experimental feature should be enabled for this user + channel = self.make_request( + "GET", + "/_matrix/client/versions", + content={}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) + self._sanity_check_versions_response(channel.json_body) + self.assertEqual( + channel.json_body["unstable_features"]["org.matrix.msc3881"], + True, + channel.json_body, + ) + + # But not for other users + channel = self.make_request( + "GET", + "/_matrix/client/versions", + content={}, + access_token=user2_tok, + ) + self.assertEqual(channel.code, 200, channel.result) + self._sanity_check_versions_response(channel.json_body) + self.assertEqual( + channel.json_body["unstable_features"]["org.matrix.msc3881"], + False, + channel.json_body, + ) + + def _sanity_check_versions_response(self, versions_response: JsonDict) -> None: + """ + Make sure this looks like a `/_matrix/client/versions` response + """ + self.assertIsInstance( + versions_response["versions"], + list, + f"Expected `versions` to be a list of strings but saw {versions_response}", + ) + self.assertIsInstance( + versions_response["unstable_features"], + dict, + f"Expected `unstable_features` to be a dict mapping feature name to a bool but saw {versions_response}", + ) + + def _enable_experimental_feature_for_user( + self, *, target_user_id: str, features: dict[str, bool] + ) -> None: + """ + Use the admin API to enable an experimental feature for a specific user + """ + channel = self.make_request( + "PUT", + f"/_synapse/admin/v1/experimental_features/{target_user_id}", + content={ + "features": features, + }, + access_token=self.admin_user_tok, + ) + self.assertEqual(channel.code, 200) diff --git a/tests/server.py b/tests/server.py index ce5eaad63da..52b7dba7491 100644 --- a/tests/server.py +++ b/tests/server.py @@ -101,6 +101,7 @@ from synapse.storage.prepare_database import prepare_database from synapse.types import ISynapseReactor, JsonDict from synapse.util.clock import Clock +from synapse.util.duration import Duration from synapse.util.json import json_encoder from tests.utils import ( @@ -301,15 +302,59 @@ def transport(self) -> "FakeChannel": def await_result(self, timeout_ms: int = 1000) -> None: """ Wait until the request is finished. + + Advances the Twisted reactor clock by 0.1s and suspending execution of the + Python thread (to allow other threads to do work) in a loop until we see a + result. We timeout when both the Twisted reactor clock has been advanced enough + AND we've waited the same amount of in real-time for the specified timeout + before giving up. + + The loop 1) allows `clock.call_later` scheduled callbacks to run if they are + scheduled to run now and 2) will also allow other threads to make progress. This + could be things spawned on the Twisted reactor threadpool or Tokio runtime + (async Rust code). """ - end_time = self._reactor.seconds() + timeout_ms / 1000.0 + timeout = Duration(milliseconds=timeout_ms) + start_time_seconds = self._reactor.seconds() + start_real_time_seconds = time.time() + + # TODO: Why? self._reactor.run() while not self.is_finished(): - if self._reactor.seconds() > end_time: + if ( + # Exceeded the Twisted reactor time timeout + start_time_seconds + timeout.as_secs() < self._reactor.seconds() + # And exceeded the real-time timeout + and start_real_time_seconds + timeout.as_secs() < time.time() + ): raise TimedOutException("Timed out waiting for request to finish.") - self._reactor.advance(0.1) + # Suspend execution of this thread to allow other threads to do work. This + # could be things spawned on the Twisted reactor threadpool or Tokio thread + # pool (async Rust code). + # + # Note: Since we're waiting real-time (`timeout` duration), the tests also + # pass with `time.sleep(0)` commented out because Python has a default + # thread switch interval (5ms for cpython) (see + # `sys.setswitchinterval(interval)`). We still want this here as we're able + # to preempt and cause the thread context swtich to happen faster. + time.sleep(0) + + # Advance the Twisted reactor and run any scheduled callbacks + # + # Don't advance the Twisted reactor clock further than the timeout duration + # as someone should increase the timeout if they expect things to take + # longer. + if start_time_seconds + timeout.as_secs() > self._reactor.seconds(): + self._reactor.advance(0.1) + else: + # But we want to still keep running whatever might be getting scheduled + # to run now. + # + # For example from other threads, they may have scheduled something on + # the reactor to run (like `reactor.callFromThread(...)`) + self._reactor.advance(0) def extract_cookies(self, cookies: MutableMapping[str, str]) -> None: """Process the contents of any Set-Cookie headers in the response From 35163d8c391189abecacaa203a110483a092f9a7 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 24 Jun 2026 19:06:41 -0500 Subject: [PATCH 06/32] Add changelog --- changelog.d/19878.misc | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/19878.misc diff --git a/changelog.d/19878.misc b/changelog.d/19878.misc new file mode 100644 index 00000000000..9ea93729c96 --- /dev/null +++ b/changelog.d/19878.misc @@ -0,0 +1 @@ +Allow Rust code to have database access via Python database connection pool. From c4717ae6f8078aeab159baa0085e61501caf8dc2 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 24 Jun 2026 19:18:27 -0500 Subject: [PATCH 07/32] Fix typo --- rust/src/storage/store.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rust/src/storage/store.rs b/rust/src/storage/store.rs index e67b25e496c..6900fda3b07 100644 --- a/rust/src/storage/store.rs +++ b/rust/src/storage/store.rs @@ -91,7 +91,7 @@ impl Store { // Otherwise, we should only find a single row for this (user, feature) [row] => Some(row.try_get(0)?), _ => { - panic!("Programming error (SQL query probably doens't match our expectations)") + panic!("Programming error (SQL query probably doesn't match our expectations)") } }; From ba0793e7e7ea921c3ffead99e169cb56c32d21b5 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 24 Jun 2026 19:19:01 -0500 Subject: [PATCH 08/32] Restore lost comment --- rust/src/deferred.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/rust/src/deferred.rs b/rust/src/deferred.rs index b915d65e078..0e61ba8c41e 100644 --- a/rust/src/deferred.rs +++ b/rust/src/deferred.rs @@ -70,6 +70,8 @@ fn logging_context_module(py: Python<'_>) -> PyResult<&Bound<'_, PyAny>> { /// Creates a twisted deferred from the given future, spawning the task on the /// tokio runtime. +/// +/// Does not handle deferred cancellation or contextvars. pub fn create_deferred<'py, F, O>( py: Python<'py>, reactor: &Bound<'py, PyAny>, From 1f08bc01f50502fac24fd58a4a7ada370c1e0cbc Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 25 Jun 2026 16:41:50 -0500 Subject: [PATCH 09/32] Backport `FakeChannel.await_result(...)` changes from #19879 See https://github.com/element-hq/synapse/pull/19879 See https://github.com/element-hq/synapse/pull/19878#discussion_r3471110281 --- tests/server.py | 38 ++++++++++++++++++++++++++++++++------ 1 file changed, 32 insertions(+), 6 deletions(-) diff --git a/tests/server.py b/tests/server.py index 52b7dba7491..2069565a147 100644 --- a/tests/server.py +++ b/tests/server.py @@ -306,27 +306,42 @@ def await_result(self, timeout_ms: int = 1000) -> None: Advances the Twisted reactor clock by 0.1s and suspending execution of the Python thread (to allow other threads to do work) in a loop until we see a result. We timeout when both the Twisted reactor clock has been advanced enough - AND we've waited the same amount of in real-time for the specified timeout - before giving up. + AND we've waited the 1s of real-time before giving up. The loop 1) allows `clock.call_later` scheduled callbacks to run if they are scheduled to run now and 2) will also allow other threads to make progress. This could be things spawned on the Twisted reactor threadpool or Tokio runtime (async Rust code). + + Args: + timeout_ms: The Twisted reactor time we wait until we raise a `TimedOutException` """ timeout = Duration(milliseconds=timeout_ms) start_time_seconds = self._reactor.seconds() + + # 1s is an arbitrary small number so we don't have to wait that long when + # something is stuck and because we assume any task on another thread will be + # fast enough. + # + # We don't use the same `timeout_ms` passed in because some tests specify 20s + # and we don't want to be waiting that long unnecessarily. + real_time_timeout = Duration(seconds=1) start_real_time_seconds = time.time() # TODO: Why? self._reactor.run() + loop_count = 0 while not self.is_finished(): if ( # Exceeded the Twisted reactor time timeout - start_time_seconds + timeout.as_secs() < self._reactor.seconds() + # + # We use `>=` for the reactor time condition as it's possible we advance + # exactly the `timeout` amount and we don't want to get stuck in an + # infinite loop + self._reactor.seconds() >= start_time_seconds + timeout.as_secs() # And exceeded the real-time timeout - and start_real_time_seconds + timeout.as_secs() < time.time() + and time.time() > start_real_time_seconds + real_time_timeout.as_secs() ): raise TimedOutException("Timed out waiting for request to finish.") @@ -339,14 +354,23 @@ def await_result(self, timeout_ms: int = 1000) -> None: # thread switch interval (5ms for cpython) (see # `sys.setswitchinterval(interval)`). We still want this here as we're able # to preempt and cause the thread context swtich to happen faster. - time.sleep(0) + # + # After a few cycles, we use `time.sleep(0.001)` instead of `time.sleep(0)` + # to avoid tightlooping on the main thread (CPU 100%) because it's wasteful + # and may starve out other threads. 10 is arbitrary but many cases will have + # none or only a few round-trips so we can just try to go as fast as + # posssible. + if loop_count < 10: + time.sleep(0) + else: + time.sleep(0.001) # Advance the Twisted reactor and run any scheduled callbacks # # Don't advance the Twisted reactor clock further than the timeout duration # as someone should increase the timeout if they expect things to take # longer. - if start_time_seconds + timeout.as_secs() > self._reactor.seconds(): + if self._reactor.seconds() < start_time_seconds + timeout.as_secs(): self._reactor.advance(0.1) else: # But we want to still keep running whatever might be getting scheduled @@ -356,6 +380,8 @@ def await_result(self, timeout_ms: int = 1000) -> None: # the reactor to run (like `reactor.callFromThread(...)`) self._reactor.advance(0) + loop_count += 1 + def extract_cookies(self, cookies: MutableMapping[str, str]) -> None: """Process the contents of any Set-Cookie headers in the response From 1dd1632ccefc500312d5390504a06efa3ed57275 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 15:55:16 -0500 Subject: [PATCH 10/32] Use struct update syntax (`..`) instead of clone See https://github.com/element-hq/synapse/pull/19878#discussion_r3491204319 --- rust/src/handlers/versions.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/rust/src/handlers/versions.rs b/rust/src/handlers/versions.rs index 97e31440271..d7ac679887d 100644 --- a/rust/src/handlers/versions.rs +++ b/rust/src/handlers/versions.rs @@ -123,8 +123,7 @@ async fn build_versions_response( let unstable_feature_map = UnstableFeatureMap { msc3575: msc3575_enabled, msc3881: msc3881_enabled, - // The clone here isn't the best but better than manually composing things - ..global_unstable_feature_map.clone() + ..*global_unstable_feature_map }; Ok(VersionsResponse { From 3256be90dc8ae5301d1ae64cbe9cfd006e439629 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 15:56:49 -0500 Subject: [PATCH 11/32] No need to inline See https://github.com/element-hq/synapse/pull/19878#discussion_r3491772557 --- rust/src/storage/db/python_db_pool.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index 28d23e57800..0464d2cddce 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/rust/src/storage/db/python_db_pool.rs @@ -51,7 +51,6 @@ pub enum DatabaseEngine { } impl DatabaseEngine { - //[inline] pub fn supports_using_any_list(&self) -> bool { match self { DatabaseEngine::Sqlite => false, From cf181ce844c98523293a46e53b5a96212f8efd4e Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 16:02:55 -0500 Subject: [PATCH 12/32] Use `FutureExt::now_or_never` instead of custom `poll_once` See https://github.com/element-hq/synapse/pull/19878#discussion_r3491831181 --- rust/src/storage/db/python_db_pool.rs | 25 +++++-------------------- 1 file changed, 5 insertions(+), 20 deletions(-) diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index 0464d2cddce..c27213d6f51 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/rust/src/storage/db/python_db_pool.rs @@ -25,12 +25,10 @@ //! - transaction [`LoggingTransactionWrapper`] (implements [`Transaction`]) and query //! the database. -use std::future::Future; -use std::pin::pin; use std::sync::{Arc, Mutex}; -use std::task::Poll; use anyhow::Context; +use futures::FutureExt; use pyo3::{ exceptions::{PyRuntimeError, PyTypeError}, intern, @@ -127,12 +125,12 @@ impl DatabasePool for PythonDatabasePoolWrapper { // Alternatively, we could just use `futures::executor::block_on` // which is probably cleaner but a single-shot poll is more // enforcing of the concept we want to represent. - match poll_once(func(&mut txn)) { - Poll::Ready(Ok(value)) => { + match func(&mut txn).now_or_never() { + Some(Ok(value)) => { *callback_slot.lock().unwrap() = Some(Ok(value)); Ok(py.None()) } - Poll::Ready(Err(err)) => { + Some(Err(err)) => { // Re-raise into Python so `runInteraction` rolls the // transaction back (and can apply its retry logic for // serialization/deadlock errors). @@ -140,7 +138,7 @@ impl DatabasePool for PythonDatabasePoolWrapper { *callback_slot.lock().unwrap() = Some(Err(err)); Err(py_err) } - Poll::Pending => unreachable!( + None => unreachable!( "The `run_interaction` transaction callback future returned `Poll::Pending`, \ but we expect Synapse Python database work to resolve synchronously. \ This is a Synapse programming error: genuine async work is \ @@ -188,19 +186,6 @@ impl DatabasePool for PythonDatabasePoolWrapper { } } -/// Poll a future exactly once. -/// -/// Returns [`Poll::Ready`] if the future resolves immediately, or [`Poll::Pending`] if -/// it would need to suspend. We use this where a future is expected to never genuinely -/// suspend and want to enforce/check that, rather than driving it to completion with a -/// blocking executor like `futures::executor::block_on`. -fn poll_once(future: F) -> Poll { - let mut future = pin!(future); - let waker = futures::task::noop_waker(); - let mut cx = std::task::Context::from_waker(&waker); - future.as_mut().poll(&mut cx) -} - /// 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 From 7e2a81424c051edb64cda08d9d20198cfb435d7d Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 16:06:09 -0500 Subject: [PATCH 13/32] Docstring for `execute` See https://github.com/element-hq/synapse/pull/19878#discussion_r3491903493 --- rust/src/storage/db/python_db_pool.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index c27213d6f51..930c794fdde 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/rust/src/storage/db/python_db_pool.rs @@ -251,6 +251,7 @@ impl<'a, 'py> FromPyObject<'a, 'py> for LoggingTransactionWrapper { } impl LoggingTransactionWrapper { + /// Calls the Python `LoggingTransaction.execute` function. fn execute<'py>( &mut self, py: Python<'py>, From 11aa532778db271ad39e8e687bf30b00082a3cd6 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 16:08:26 -0500 Subject: [PATCH 14/32] Use raw string for SQL (for copy/pastability) See https://github.com/element-hq/synapse/pull/19878#discussion_r3491952013 --- rust/src/storage/store.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/rust/src/storage/store.rs b/rust/src/storage/store.rs index 6900fda3b07..475a3b1db9a 100644 --- a/rust/src/storage/store.rs +++ b/rust/src/storage/store.rs @@ -78,9 +78,11 @@ impl Store { async move { let rows = txn .query( - "SELECT enabled \ - FROM per_user_experimental_features \ - WHERE user_id = ? AND feature = ?", + r#" + SELECT enabled + FROM per_user_experimental_features + WHERE user_id = ? AND feature = ? + "#, &[user_id.as_ref(), feature.as_ref()], ) .await?; From 36858c9f68a6aa35a1d844aed6e1dcb8f89fc957 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 16:10:57 -0500 Subject: [PATCH 15/32] `GET` requests don't have request bodies. See https://github.com/element-hq/synapse/pull/19878#discussion_r3492100383 --- tests/rest/client/test_versions.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/tests/rest/client/test_versions.py b/tests/rest/client/test_versions.py index 429250e58e5..d6560984690 100644 --- a/tests/rest/client/test_versions.py +++ b/tests/rest/client/test_versions.py @@ -73,7 +73,6 @@ def test_unauthenticated(self) -> None: channel = self.make_request( "GET", "/_matrix/client/versions", - content={}, ) self.assertEqual(channel.code, 200, channel.result) self._sanity_check_versions_response(channel.json_body) @@ -85,7 +84,6 @@ def test_authenticated(self) -> None: channel = self.make_request( "GET", "/_matrix/client/versions", - content={}, access_token=user1_tok, ) self.assertEqual(channel.code, 200, channel.result) @@ -101,7 +99,6 @@ def test_authenticated_with_per_user_feature(self) -> None: channel = self.make_request( "GET", "/_matrix/client/versions", - content={}, access_token=user1_tok, ) self.assertEqual(channel.code, 200, channel.result) @@ -121,7 +118,6 @@ def test_authenticated_with_per_user_feature(self) -> None: channel = self.make_request( "GET", "/_matrix/client/versions", - content={}, access_token=user1_tok, ) self.assertEqual(channel.code, 200, channel.result) @@ -136,7 +132,6 @@ def test_authenticated_with_per_user_feature(self) -> None: channel = self.make_request( "GET", "/_matrix/client/versions", - content={}, access_token=user2_tok, ) self.assertEqual(channel.code, 200, channel.result) From 2233a25d29b9e598d9788309479c0ba3c45fd8b2 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 16:12:47 -0500 Subject: [PATCH 16/32] We can use `UnstableFeatureMap` directly instead of unnecessary translation to `std::collections::BTreeMap` See https://github.com/element-hq/synapse/pull/19878#discussion_r3492618490 --- rust/src/handlers/versions.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/rust/src/handlers/versions.rs b/rust/src/handlers/versions.rs index d7ac679887d..a962cc34edd 100644 --- a/rust/src/handlers/versions.rs +++ b/rust/src/handlers/versions.rs @@ -17,18 +17,18 @@ use std::sync::Arc; use pyo3::prelude::*; use pythonize::{pythonize, PythonizeError}; -use serde::{Deserialize, Serialize}; +use serde::Serialize; use crate::config::{RoomCreationPreset, SynapseHomeServerConfig}; use crate::deferred::create_deferred; use crate::storage::store::{PerUserExperimentalFeature, Store}; /// `GET /_matrix/client/versions` response -#[derive(Serialize, Deserialize, Clone, Debug)] +#[derive(Serialize, Clone, Debug)] struct VersionsResponse { versions: Vec, /// as per MSC1497 - unstable_features: std::collections::BTreeMap, + unstable_features: UnstableFeatureMap, } impl<'py> IntoPyObject<'py> for VersionsResponse { @@ -156,7 +156,7 @@ async fn build_versions_response( "v1.11".to_string(), "v1.12".to_string(), ]), - unstable_features: serde_json::from_value(serde_json::to_value(unstable_feature_map)?)?, + unstable_features: unstable_feature_map, }) } From 3e278db4b225b5be8e780e6db5e3e1775468b47b Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 16:20:53 -0500 Subject: [PATCH 17/32] No need to explain fundamental PyO3 concepts See https://github.com/element-hq/synapse/pull/19878#discussion_r3491886556 --- rust/src/storage/db/python_db_pool.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index 930c794fdde..cf65506b116 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/rust/src/storage/db/python_db_pool.rs @@ -225,9 +225,7 @@ pub struct LoggingTransactionWrapper { /// The underlying `LoggingTransaction` /// /// We purposely avoid `Bound<'py, PyAny>` so it can be stored and moved freely - /// across threads (like to extract it from the `new_transaction(...)` callback). - /// You will need to acquire your own `py` and bind it using - /// `logging_transaction_py.bind(py)` to do anything useful. + /// across threads (as required by `Transaction` trait). logging_transaction_py: Py, /// Disambiguate which underlying database engine we're working with From 12a6209540c695cbb709f4a2e16a3dbe4c5fb101 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 16:35:50 -0500 Subject: [PATCH 18/32] Error instead of panic for incorrect number of rows from SQL query See https://github.com/element-hq/synapse/pull/19878#discussion_r3492036656 --- rust/src/storage/store.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/rust/src/storage/store.rs b/rust/src/storage/store.rs index 475a3b1db9a..7b5feada7b1 100644 --- a/rust/src/storage/store.rs +++ b/rust/src/storage/store.rs @@ -92,8 +92,12 @@ impl Store { [] => None, // Otherwise, we should only find a single row for this (user, feature) [row] => Some(row.try_get(0)?), - _ => { - panic!("Programming error (SQL query probably doesn't match our expectations)") + rows => { + anyhow::bail!( + "Unexpected number of rows returned (expected exactly 0 or 1, saw {}). \ + This probably means the SQL query probably doesn't match our expectations.", + rows.len(), + ); } }; From 8c320343e073548a73c773a6a6b19655ae6af42a Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 16:57:26 -0500 Subject: [PATCH 19/32] Split off `RoomCreationPreset` to its own file So we can stop muddying up the actual config --- rust/src/config/mod.rs | 46 +++++++----------------------------- rust/src/config/types.rs | 51 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 38 deletions(-) create mode 100644 rust/src/config/types.rs diff --git a/rust/src/config/mod.rs b/rust/src/config/mod.rs index 47981d5de00..84e37f45c33 100644 --- a/rust/src/config/mod.rs +++ b/rust/src/config/mod.rs @@ -14,17 +14,20 @@ */ use std::collections::BTreeSet; -use std::str::FromStr; -use pyo3::{exceptions::PyRuntimeError, prelude::*}; +use pyo3::prelude::*; + +pub mod types; /// A Rust-side view of Synapse's Python `HomeServerConfig`. /// /// This only mirrors the subset of config that the Rust handlers need, rather /// than the whole thing. Thanks to `#[derive(FromPyObject)]`, each field is /// pulled directly off the corresponding attribute of the Python `config` -/// object, so you can populate it in one shot with -/// `homeserver.getattr("config")?.extract()?`. +/// object, so you can populate it in one shot with: +/// ``` +/// let config: SynapseHomeServerConfig = homeserver.getattr("config")?.extract()?; +/// ``` #[derive(FromPyObject, Clone)] pub struct SynapseHomeServerConfig { pub room: RoomConfig, @@ -33,42 +36,9 @@ pub struct SynapseHomeServerConfig { pub experimental: ExperimentalConfig, } -#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] -pub enum RoomCreationPreset { - PrivateChat, - PublicChat, - TrustedPrivateChat, -} - -impl FromStr for RoomCreationPreset { - type Err = PyErr; - - fn from_str(s: &str) -> Result { - Ok(match s { - "private_chat" => RoomCreationPreset::PrivateChat, - "public_chat" => RoomCreationPreset::PublicChat, - "trusted_private_chat" => RoomCreationPreset::TrustedPrivateChat, - other => { - return Err(PyRuntimeError::new_err(format!( - "Unknown variant {other:?} does not translate to `RoomCreationPreset`. \ - This is a Synapse programming error." - ))) - } - }) - } -} - -impl<'a, 'py> FromPyObject<'a, 'py> for RoomCreationPreset { - type Error = PyErr; - - fn extract(value: Borrowed<'a, 'py, PyAny>) -> PyResult { - value.extract::<&str>()?.parse() - } -} - #[derive(FromPyObject, Clone)] pub struct RoomConfig { - pub encryption_enabled_by_default_for_room_presets: BTreeSet, + pub encryption_enabled_by_default_for_room_presets: BTreeSet, } #[derive(FromPyObject, Clone)] diff --git a/rust/src/config/types.rs b/rust/src/config/types.rs new file mode 100644 index 00000000000..c6cb79caa75 --- /dev/null +++ b/rust/src/config/types.rs @@ -0,0 +1,51 @@ +/* + * 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: + * . + * + */ + +use std::str::FromStr; + +use pyo3::{exceptions::PyRuntimeError, prelude::*}; + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub enum RoomCreationPreset { + PrivateChat, + PublicChat, + TrustedPrivateChat, +} + +impl FromStr for RoomCreationPreset { + type Err = PyErr; + + fn from_str(s: &str) -> Result { + Ok(match s { + "private_chat" => RoomCreationPreset::PrivateChat, + "public_chat" => RoomCreationPreset::PublicChat, + "trusted_private_chat" => RoomCreationPreset::TrustedPrivateChat, + other => { + return Err(PyRuntimeError::new_err(format!( + "Unknown variant {other:?} does not translate to `RoomCreationPreset`. \ + This is a Synapse programming error." + ))) + } + }) + } +} + +impl<'a, 'py> FromPyObject<'a, 'py> for RoomCreationPreset { + type Error = PyErr; + + fn extract(value: Borrowed<'a, 'py, PyAny>) -> PyResult { + value.extract::<&str>()?.parse() + } +} From 826782211981b78474221bcc2fcc4909e9aafcfb Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Thu, 2 Jul 2026 16:58:17 -0500 Subject: [PATCH 20/32] Add docstring to `RoomCreationPreset` --- rust/src/config/types.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/rust/src/config/types.rs b/rust/src/config/types.rs index c6cb79caa75..2f05be4fa4a 100644 --- a/rust/src/config/types.rs +++ b/rust/src/config/types.rs @@ -17,6 +17,15 @@ use std::str::FromStr; use pyo3::{exceptions::PyRuntimeError, prelude::*}; +/// The presets available when creating a Matrix room according to the Matrix spec: +/// +/// > Preset | join_rules | history_visibility | guest_access | Other +/// > --- | --- | --- | --- | --- +/// > `private_chat` | `invite` | ``shared`` | `can_join` | . +/// > `trusted_private_chat` | `invite` | shared | `can_join` | All invitees are given the same power level as the room creator. +/// > `public_chat` | `public` | `shared` | `forbidden` | . +/// > +/// > *-- https://spec.matrix.org/v1.18/client-server-api/#post_matrixclientv3createroom* #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] pub enum RoomCreationPreset { PrivateChat, From 962c0fbb4afc488a8fa143241783e7fcf01e1f5e Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 11:05:53 -0500 Subject: [PATCH 21/32] Fix lint --- rust/src/handlers/versions.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rust/src/handlers/versions.rs b/rust/src/handlers/versions.rs index a962cc34edd..db8bca82d68 100644 --- a/rust/src/handlers/versions.rs +++ b/rust/src/handlers/versions.rs @@ -19,7 +19,7 @@ use pyo3::prelude::*; use pythonize::{pythonize, PythonizeError}; use serde::Serialize; -use crate::config::{RoomCreationPreset, SynapseHomeServerConfig}; +use crate::config::{types::RoomCreationPreset, SynapseHomeServerConfig}; use crate::deferred::create_deferred; use crate::storage::store::{PerUserExperimentalFeature, Store}; From 4fa74e590a57ba852b80bc29acb1dd144ed1dbae Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 11:13:04 -0500 Subject: [PATCH 22/32] Remove old comment --- rust/src/storage/store.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/rust/src/storage/store.rs b/rust/src/storage/store.rs index 7b5feada7b1..b339d7748c5 100644 --- a/rust/src/storage/store.rs +++ b/rust/src/storage/store.rs @@ -61,8 +61,6 @@ impl Store { user_id: &str, feature: PerUserExperimentalFeature, ) -> Result, anyhow::Error> { - // It's not enabled globally, so check whether it's enabled per-user. - // // We need owned copies to move into the callback because it is `'static` (it // may be moved to another thread). We use `Arc` rather than `String` so // the per-call clone is just a cheap refcount bump rather than a fresh From 22a506454c46ce120c87e953f61a3d9819542064 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 12:33:06 -0500 Subject: [PATCH 23/32] Be more explicit about valid values for bool --- rust/src/storage/db/mod.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/rust/src/storage/db/mod.rs b/rust/src/storage/db/mod.rs index 126faf15eab..c75c206a389 100644 --- a/rust/src/storage/db/mod.rs +++ b/rust/src/storage/db/mod.rs @@ -207,7 +207,11 @@ impl FromDbValue for bool { match value { DbValue::Bool(b) => Ok(b), // SQLite has no native boolean type and stores them as integers. - DbValue::Int(i) => Ok(i != 0), + DbValue::Int(i) => match i { + 0 => Ok(false), + 1 => Ok(true), + _ => anyhow::bail!("cannot read DbValue::Int({i}) as bool"), + }, other => anyhow::bail!("cannot read {other:?} as bool"), } } From 95f0d155b625dd0cb5133351a8994f358e98d3fb Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 13:56:34 -0500 Subject: [PATCH 24/32] Remove sketchy extra `as` casts --- rust/src/storage/db/mod.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/rust/src/storage/db/mod.rs b/rust/src/storage/db/mod.rs index c75c206a389..d2bbd0135b8 100644 --- a/rust/src/storage/db/mod.rs +++ b/rust/src/storage/db/mod.rs @@ -221,7 +221,6 @@ impl FromDbValue for i64 { fn from_value(value: DbValue) -> Result { match value { DbValue::Int(i) => Ok(i), - DbValue::Bool(b) => Ok(b as i64), other => anyhow::bail!("cannot read {other:?} as i64"), } } @@ -231,7 +230,6 @@ impl FromDbValue for f64 { fn from_value(value: DbValue) -> Result { match value { DbValue::Float(f) => Ok(f), - DbValue::Int(i) => Ok(i as f64), other => anyhow::bail!("cannot read {other:?} as f64"), } } From 49afcb88342bcdde0e837107772872ee8f735ef3 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 14:27:23 -0500 Subject: [PATCH 25/32] Compare actual Python types for `detect_engine(...)` See https://github.com/element-hq/synapse/pull/19878#discussion_r3491877766 --- rust/src/storage/db/python_db_pool.rs | 68 ++++++++++++++++++++------- 1 file changed, 51 insertions(+), 17 deletions(-) diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index cf65506b116..c8f704537a2 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/rust/src/storage/db/python_db_pool.rs @@ -29,6 +29,7 @@ use std::sync::{Arc, Mutex}; use anyhow::Context; use futures::FutureExt; +use once_cell::sync::OnceCell; use pyo3::{ exceptions::{PyRuntimeError, PyTypeError}, intern, @@ -41,6 +42,41 @@ use crate::storage::db::{ DatabasePool, DbRow, DbValue, ErasedInteraction, ErasedResult, Transaction, }; +/// A reference to the `synapse.storage.engines` module. +static STORAGE_ENGINES_MODULE: OnceCell> = OnceCell::new(); + +/// Access to the `synapse.storage.engines` module. +fn storage_engines_module(py: Python<'_>) -> PyResult<&Bound<'_, PyAny>> { + Ok(STORAGE_ENGINES_MODULE + .get_or_try_init(|| py.import("synapse.storage.engines").map(Into::into))? + .bind(py)) +} + +static SQLITE3_ENGINE_CLASS: OnceCell> = OnceCell::new(); +static POSTGRES_ENGINE_CLASS: OnceCell> = OnceCell::new(); + +/// Access to the `Sqlite3Engine` class +fn sqlite3_engine_class(py: Python<'_>) -> PyResult<&Bound<'_, PyAny>> { + Ok(SQLITE3_ENGINE_CLASS + .get_or_try_init(|| { + storage_engines_module(py)? + .getattr("Sqlite3Engine") + .map(Into::into) + })? + .bind(py)) +} + +/// Access to the `PostgresEngine` class +fn postgres_engine_class(py: Python<'_>) -> PyResult<&Bound<'_, PyAny>> { + Ok(POSTGRES_ENGINE_CLASS + .get_or_try_init(|| { + storage_engines_module(py)? + .getattr("PostgresEngine") + .map(Into::into) + })? + .bind(py)) +} + /// The database engines we support in the Python side of Synapse #[derive(Copy, Clone, Debug)] pub enum DatabaseEngine { @@ -201,23 +237,21 @@ fn anyhow_to_pyerr(err: &anyhow::Error) -> PyErr { /// Given a Python `LoggingTransaction`, figures out the database engine that backs it fn detect_engine(txn_py: &Bound<'_, PyAny>) -> PyResult { - let name = txn_py - .getattr("database_engine") - .expect("`LoggingTransaction` must have `database_engine` attr") - .get_type() - .name() - .expect("`database_engine` type must have a name") - .to_str() - .expect("`database_engine` type name must be valid UTF-8") - .to_owned(); - - Ok(match name.as_str() { - "PostgresEngine" => DatabaseEngine::Postgres, - "Sqlite3Engine" => DatabaseEngine::Sqlite, - other => unimplemented!( - "Unknown database engine {other:?}. This is a Synapse programming error." - ), - }) + let py = txn_py.py(); + let database_engine = txn_py.getattr(intern!(py, "database_engine"))?; + + // Compare against the actual engine classes imported from Python (the PyO3 + // equivalent of an `isinstance` check). + if database_engine.is_instance(postgres_engine_class(py)?)? { + Ok(DatabaseEngine::Postgres) + } else if database_engine.is_instance(sqlite3_engine_class(py)?)? { + Ok(DatabaseEngine::Sqlite) + } else { + Err(PyTypeError::new_err(format!( + "Unknown database engine {}. This is a Synapse programming error.", + database_engine.get_type().name()? + ))) + } } /// Wrapper for a `LoggingTransaction` from the Python side of Synapse. From 5b57d5c76bf0e359772baa8579ae6abcd44eaab0 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 15:03:15 -0500 Subject: [PATCH 26/32] Remove `unwrap` usage from Python `run_interaction_erased` See https://github.com/element-hq/synapse/pull/19878#discussion_r3491836917 --- rust/src/storage/db/python_db_pool.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index c8f704537a2..ecca0e0cd60 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/rust/src/storage/db/python_db_pool.rs @@ -163,7 +163,10 @@ impl DatabasePool for PythonDatabasePoolWrapper { // enforcing of the concept we want to represent. match func(&mut txn).now_or_never() { Some(Ok(value)) => { - *callback_slot.lock().unwrap() = Some(Ok(value)); + let mut callback_slot = callback_slot + .lock() + .map_err(|err| anyhow::anyhow!("Failed to acquire lock on `callback_slot`: {:#}", err))?; + *callback_slot = Some(Ok(value)); Ok(py.None()) } Some(Err(err)) => { @@ -171,7 +174,10 @@ impl DatabasePool for PythonDatabasePoolWrapper { // 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)); + let mut callback_slot = callback_slot + .lock() + .map_err(|err| anyhow::anyhow!("Failed to acquire lock on `callback_slot`: {:#}", err))?; + *callback_slot = Some(Err(err)); Err(py_err) } None => unreachable!( From 936a77da5f00c80895591f00b31e797e7d947f1e Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 16:17:31 -0500 Subject: [PATCH 27/32] Remove `result_slot` `unwrap` --- rust/src/storage/db/python_db_pool.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index ecca0e0cd60..2ad1e8c75eb 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/rust/src/storage/db/python_db_pool.rs @@ -208,7 +208,10 @@ impl DatabasePool for PythonDatabasePoolWrapper { .await; // Return the result we captured based on if `runInteraction` was successful - let captured_result = result_slot.lock().unwrap().take(); + let captured_result = result_slot + .lock() + .map_err(|err| anyhow::anyhow!("Failed to acquire lock on `result_slot`: {:#}", err))? + .take(); match run_interaction_outcome { // Only return the `captured_result` if `runInteraction` succeeded. We don't // want to accidentally return a successful result when the transaction From 3cd3fd8fc35224694baa77b3e3e92b30f52696bb Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 16:18:36 -0500 Subject: [PATCH 28/32] Remove `unwrap` usage with `success_sender`/`error_sender` --- rust/src/deferred.rs | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/rust/src/deferred.rs b/rust/src/deferred.rs index 0e61ba8c41e..62a1ce6b903 100644 --- a/rust/src/deferred.rs +++ b/rust/src/deferred.rs @@ -158,7 +158,13 @@ where None, move |args, _kwargs| -> PyResult> { let value = args.get_item(0)?.unbind(); - if let Some(tx) = success_sender.lock().unwrap().take() { + if let Some(tx) = success_sender + .lock() + .map_err(|err| { + anyhow::anyhow!("Failed to acquire lock on `success_sender`: {:#}", err) + })? + .take() + { let _ = tx.send(Ok(value)); } Ok(args.py().None()) @@ -173,7 +179,13 @@ where None, move |args, _kwargs| -> PyResult> { let err = failure_to_pyerr(&args.get_item(0)?); - if let Some(tx) = error_sender.lock().unwrap().take() { + if let Some(tx) = error_sender + .lock() + .map_err(|err| { + anyhow::anyhow!("Failed to acquire lock on `error_sender`: {:#}", err) + })? + .take() + { let _ = tx.send(Err(err)); } Ok(args.py().None()) From 98b43cdaf97856a4d73ce79ea4eeb5ad7a572407 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 16:36:55 -0500 Subject: [PATCH 29/32] Remove `expect` from `run_interaction` --- rust/src/storage/db/mod.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/rust/src/storage/db/mod.rs b/rust/src/storage/db/mod.rs index d2bbd0135b8..c6a5ecd8fba 100644 --- a/rust/src/storage/db/mod.rs +++ b/rust/src/storage/db/mod.rs @@ -125,9 +125,11 @@ pub trait DatabasePoolExt: DatabasePool { 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)", - )) + boxed.downcast::().map(|b| *b).map_err(|_| { + anyhow::anyhow!( + "run_interaction return type mismatch (this is a Synapse programming error)" + ) + }) } } } From 47fb4f910a4333b46abdcd8c8aafbed957c5a164 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 17:04:18 -0500 Subject: [PATCH 30/32] Remove `unreachable!` panic for unexpected async work on Python DB pool --- rust/src/storage/db/python_db_pool.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index 2ad1e8c75eb..2c206a64127 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/rust/src/storage/db/python_db_pool.rs @@ -180,12 +180,14 @@ impl DatabasePool for PythonDatabasePoolWrapper { *callback_slot = Some(Err(err)); Err(py_err) } - None => unreachable!( - "The `run_interaction` transaction callback future returned `Poll::Pending`, \ - but we expect Synapse Python database work to resolve synchronously. \ - This is a Synapse programming error: genuine async work is \ - not supported here.", - ), + None => { + Err(PyRuntimeError::new_err( + "The `run_interaction` transaction callback future returned `Poll::Pending`, \ + but we expect Synapse Python database work to resolve synchronously. \ + This is a Synapse programming error: genuine async work is \ + not supported here.", + )) + } } }, )? From 930fb59b5dfdbe5102ecca38255996af4ff71ada Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Fri, 3 Jul 2026 17:50:52 -0500 Subject: [PATCH 31/32] Use `PyAssertionError` for programming errors --- rust/src/config/types.rs | 4 ++-- rust/src/storage/db/python_db_pool.rs | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/rust/src/config/types.rs b/rust/src/config/types.rs index 2f05be4fa4a..8990b52f656 100644 --- a/rust/src/config/types.rs +++ b/rust/src/config/types.rs @@ -15,7 +15,7 @@ use std::str::FromStr; -use pyo3::{exceptions::PyRuntimeError, prelude::*}; +use pyo3::{exceptions::PyAssertionError, prelude::*}; /// The presets available when creating a Matrix room according to the Matrix spec: /// @@ -42,7 +42,7 @@ impl FromStr for RoomCreationPreset { "public_chat" => RoomCreationPreset::PublicChat, "trusted_private_chat" => RoomCreationPreset::TrustedPrivateChat, other => { - return Err(PyRuntimeError::new_err(format!( + return Err(PyAssertionError::new_err(format!( "Unknown variant {other:?} does not translate to `RoomCreationPreset`. \ This is a Synapse programming error." ))) diff --git a/rust/src/storage/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs index 2c206a64127..5e62b066561 100644 --- a/rust/src/storage/db/python_db_pool.rs +++ b/rust/src/storage/db/python_db_pool.rs @@ -31,7 +31,7 @@ use anyhow::Context; use futures::FutureExt; use once_cell::sync::OnceCell; use pyo3::{ - exceptions::{PyRuntimeError, PyTypeError}, + exceptions::{PyAssertionError, PyRuntimeError, PyTypeError}, intern, prelude::*, types::{PyBool, PyCFunction, PyFloat, PyInt, PyList, PyString}, @@ -181,7 +181,7 @@ impl DatabasePool for PythonDatabasePoolWrapper { Err(py_err) } None => { - Err(PyRuntimeError::new_err( + Err(PyAssertionError::new_err( "The `run_interaction` transaction callback future returned `Poll::Pending`, \ but we expect Synapse Python database work to resolve synchronously. \ This is a Synapse programming error: genuine async work is \ From c5fe1fe53c6efb9367b2ed0c0f2d0371c63e5a6f Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Mon, 6 Jul 2026 08:05:19 -0500 Subject: [PATCH 32/32] `no_run` examples --- rust/src/config/mod.rs | 2 +- rust/src/storage/db/mod.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/rust/src/config/mod.rs b/rust/src/config/mod.rs index 84e37f45c33..62370571279 100644 --- a/rust/src/config/mod.rs +++ b/rust/src/config/mod.rs @@ -25,7 +25,7 @@ pub mod types; /// than the whole thing. Thanks to `#[derive(FromPyObject)]`, each field is /// pulled directly off the corresponding attribute of the Python `config` /// object, so you can populate it in one shot with: -/// ``` +/// ```no_run /// let config: SynapseHomeServerConfig = homeserver.getattr("config")?.extract()?; /// ``` #[derive(FromPyObject, Clone)] diff --git a/rust/src/storage/db/mod.rs b/rust/src/storage/db/mod.rs index c6a5ecd8fba..69d9d155a32 100644 --- a/rust/src/storage/db/mod.rs +++ b/rust/src/storage/db/mod.rs @@ -79,7 +79,7 @@ pub trait DatabasePoolExt: DatabasePool { /// unnecessary waiting in your transaction anyway. /// /// Usage: - /// ```rust + /// ```no_run /// db_pool /// .run_interaction(|txn| { /// async move {