Concept: Rust database access via Python database connection pool (via runInteraction) (LLM splat v1)

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"
@@ -64,6 +65,12 @@ tokio = { version = "1.44.2", features = ["rt", "rt-multi-thread"] }
 once_cell = "1.18.0"
 itertools = "0.14.0"
 
+# TODO: Remove: These are just used to make sure a tokio-postgres backed database pool makes sense
+# with our interfaces
+bb8 = "0.8.3"
+bb8-postgres = "0.8.1"
+postgres-native-tls = "0.5.0"
+
 [build-dependencies]
 blake2 = "0.10.4"
 hex = "0.4.3"

rust/src/config/mod.rs

@@ -0,0 +1,36 @@
+/*
+ * This file is licensed under the Affero General Public License (AGPL) version 3.
+ *
+ * Copyright (C) 2026 Element Creations Ltd
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * See the GNU Affero General Public License for more details:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ *
+ */
+
+use pyo3::prelude::*;
+
+#[derive(FromPyObject, Clone)]
+pub struct SynapseConfig {
+    pub experimental: ExperimentalConfig,
+}
+
+// #[derive(FromPyObject)]
+// #[serde(rename_all = "snake_case")]
+// pub enum RoomCreationPreset {
+//     PrviateChat,
+//     PublicChat,
+//     TrustedPrivateChat,
+// }
+
+#[derive(FromPyObject, Clone)]
+pub struct ExperimentalConfig {
+    pub msc3881_enabled: bool,
+    pub msc3575_enabled: bool,
+    pub msc4222_enabled: bool,
+}

rust/src/handlers/mod.rs

@@ -0,0 +1,92 @@
+/*
+ * This file is licensed under the Affero General Public License (AGPL) version 3.
+ *
+ * Copyright (C) 2026 Element Creations Ltd
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * See the GNU Affero General Public License for more details:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ *
+ */
+
+use std::sync::Arc;
+
+use pyo3::{
+    prelude::*,
+    types::{PyAnyMethods, PyModule, PyModuleMethods},
+    Bound, PyResult, Python,
+};
+
+use crate::config::SynapseConfig;
+use crate::storage::db::python_db_pool::PythonDatabasePoolWrapper;
+use crate::storage::store::Store;
+
+pub mod versions;
+
+#[pyclass]
+struct RustHandlers {
+    versions: Py<versions::VersionsHandler>,
+}
+
+#[pymethods]
+impl RustHandlers {
+    #[new]
+    #[pyo3(signature = (homeserver))]
+    pub fn py_new(py: Python<'_>, homeserver: &Bound<'_, PyAny>) -> PyResult<RustHandlers> {
+        let config: SynapseConfig = homeserver.getattr("config")?.extract()?;
+
+        // The Twisted reactor, used both to drive our Tokio runtime and to
+        // marshal database work back onto the reactor thread.
+        let reactor: Py<PyAny> = homeserver.call_method0("get_reactor")?.unbind();
+
+        // hs.get_datastores().main.db_pool
+        let db_pool_py: Py<PyAny> = 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 {
+            config: config.clone(),
+            db_pool: Box::new(db_pool),
+        });
+
+        let versions = Py::new(
+            py,
+            versions::VersionsHandler {
+                config: config.clone(),
+                store: Arc::clone(&store),
+                reactor: reactor.clone_ref(py),
+            },
+        )?;
+
+        Ok(RustHandlers { versions })
+    }
+
+    #[getter]
+    fn versions(&self, py: Python<'_>) -> Py<versions::VersionsHandler> {
+        self.versions.clone_ref(py)
+    }
+}
+
+/// 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::<RustHandlers>()?;
+
+    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(())
+}

rust/src/handlers/versions.rs

@@ -0,0 +1,223 @@
+/*
+ * This file is licensed under the Affero General Public License (AGPL) version 3.
+ *
+ * Copyright (C) 2026 Element Creations Ltd
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * See the GNU Affero General Public License for more details:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ *
+ */
+
+use std::sync::Arc;
+
+use pyo3::prelude::*;
+use pythonize::{pythonize, PythonizeError};
+use serde::{Deserialize, Serialize};
+
+use crate::config::SynapseConfig;
+use crate::http_client::create_deferred;
+use crate::storage::store::{PerUserExperimentalFeature, Store};
+
+/// `GET /_matrix/client/versions` response
+#[derive(Serialize, Deserialize, Clone, Debug)]
+struct VersionsResponse {
+    versions: Vec<String>,
+    /// as per MSC1497
+    unstable_features: std::collections::BTreeMap<String, bool>,
+}
+
+impl<'py> IntoPyObject<'py> for VersionsResponse {
+    type Target = PyAny;
+    type Output = Bound<'py, Self::Target>;
+    type Error = PythonizeError;
+
+    fn into_pyobject(self, py: Python<'py>) -> Result<Self::Output, Self::Error> {
+        pythonize(py, &self)
+    }
+}
+
+#[pyclass]
+pub struct VersionsHandler {
+    pub config: SynapseConfig,
+    pub store: Arc<Store>,
+    /// The Twisted reactor, used to bridge our `async` response back into a
+    /// Twisted deferred that Python can `await`.
+    pub reactor: Py<PyAny>,
+}
+
+#[pymethods]
+impl VersionsHandler {
+    /// Assemble a `/versions` response, returning a Twisted deferred that
+    /// resolves to the response body (a dict).
+    #[pyo3(signature = (user_id=None))]
+    fn get_versions<'py>(
+        &self,
+        py: Python<'py>,
+        user_id: Option<String>,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let store = Arc::clone(&self.store);
+        let config = self.config.clone();
+
+        create_deferred(py, self.reactor.bind(py), async move {
+            build_versions_response(&store, &config, user_id.as_deref())
+                .await
+                .map_err(|err| {
+                    pyo3::exceptions::PyRuntimeError::new_err(format!(
+                        "Failed to build /versions response: {err:#}"
+                    ))
+                })
+        })
+    }
+}
+
+/// Assemble a `/versions` response body.
+async fn build_versions_response(
+    store: &Store,
+    config: &SynapseConfig,
+    user_id: Option<&str>,
+) -> Result<VersionsResponse, anyhow::Error> {
+    {
+        let msc3881_enabled = match user_id {
+            Some(user_id) => {
+                store
+                    .is_feature_enabled(user_id, PerUserExperimentalFeature::MSC3881)
+                    .await?
+            }
+            None => PerUserExperimentalFeature::MSC3881.is_globally_enabled(config),
+        };
+
+        let msc3575_enabled = match user_id {
+            Some(user_id) => {
+                store
+                    .is_feature_enabled(user_id, PerUserExperimentalFeature::MSC3575)
+                    .await?
+            }
+            None => PerUserExperimentalFeature::MSC3575.is_globally_enabled(config),
+        };
+
+        // TODO: Calculate these once since they shouldn't change after start-up.
+        // e2ee_forced_public = (
+        //     RoomCreationPreset.PUBLIC_CHAT
+        //     in config.room.encryption_enabled_by_default_for_room_presets
+        // );
+        // e2ee_forced_private = (
+        //     RoomCreationPreset.PRIVATE_CHAT
+        //     in config.room.encryption_enabled_by_default_for_room_presets
+        // );
+        // e2ee_forced_trusted_private = (
+        //     RoomCreationPreset.TRUSTED_PRIVATE_CHAT
+        //     in config.room.encryption_enabled_by_default_for_room_presets
+        // );
+
+        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: std::collections::BTreeMap::from([
+                // // Implements support for label-based filtering as described in
+                // // MSC2326.
+                // ("org.matrix.label_based_filtering".to_string(), true),
+                // // Implements support for cross signing as described in MSC1756
+                // ("org.matrix.e2e_cross_signing".to_string(), true),
+                // // Implements additional endpoints as described in MSC2432
+                // ("org.matrix.msc2432".to_string(), true),
+                // // Implements additional endpoints as described in MSC2666
+                // ("uk.half-shot.msc2666.query_mutual_rooms.stable".to_string(), true),
+                // // Whether new rooms will be set to encrypted or not (based on presets).
+                // ("io.element.e2ee_forced.public".to_string(), e2ee_forced_public),
+                // ("io.element.e2ee_forced.private".to_string(), e2ee_forced_private),
+                // ("io.element.e2ee_forced.trusted_private".to_string(), e2ee_forced_trusted_private),
+                // // Supports the busy presence state described in MSC3026.
+                // ("org.matrix.msc3026.busy_presence".to_string(), config.experimental.msc3026_enabled),
+                // // Supports receiving private read receipts as per MSC2285
+                // ("org.matrix.msc2285.stable".to_string(), 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".to_string(), true),
+                // // Adds support for thread relations, per MSC3440.
+                // ("org.matrix.msc3440.stable".to_string(), true),  // TODO: remove when "v1.3" is added above
+                // // Support for thread read receipts & notification counts.
+                // ("org.matrix.msc3771".to_string(), true),
+                // ("org.matrix.msc3773".to_string(), config.experimental.msc3773_enabled),
+                // // Allows moderators to fetch redacted event content as described in MSC2815
+                // ("fi.mau.msc2815".to_string(), config.experimental.msc2815_enabled),
+                // // Adds a ping endpoint for appservices to check HS->AS connection
+                // ("fi.mau.msc2659.stable".to_string(), 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".to_string(), config.auth.login_via_existing_enabled),
+                // Adds support for remotely enabling/disabling pushers, as per MSC3881
+                ("org.matrix.msc3881".to_string(), msc3881_enabled),
+                // // Adds support for filtering /messages by event relation.
+                // ("org.matrix.msc3874".to_string(), config.experimental.msc3874_enabled),
+                // // Adds support for relation-based redactions as per MSC3912.
+                // ("org.matrix.msc3912".to_string(), 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".to_string(), true),
+                // // Adds support for deleting account data.
+                // ("org.matrix.msc3391".to_string(), config.experimental.msc3391_enabled),
+                // // Allows clients to inhibit profile update propagation.
+                // ("org.matrix.msc4069".to_string(), config.experimental.msc4069_profile_inhibit_propagation),
+                // // Allows clients to handle push for encrypted events.
+                // ("org.matrix.msc4028".to_string(), 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".to_string(), (
+                //     config.experimental.msc4108_enabled
+                //     or (
+                //         config.experimental.msc4108_delegation_endpoint
+                //         is not None
+                //     )
+                // )),
+                // // MSC4140: Delayed events
+                // ("org.matrix.msc4140".to_string(), bool(config.server.max_event_delay_ms)),
+                // Simplified sliding sync
+                ("org.matrix.simplified_msc3575".to_string(), msc3575_enabled),
+                // // Arbitrary key-value profile fields.
+                // ("uk.tcpip.msc4133".to_string(), config.experimental.msc4133_enabled),
+                // ("uk.tcpip.msc4133.stable".to_string(), true),
+                // // MSC4155: Invite filtering
+                // ("org.matrix.msc4155".to_string(), config.experimental.msc4155_enabled),
+                // // MSC4306: Support for thread subscriptions
+                // ("org.matrix.msc4306".to_string(), config.experimental.msc4306_enabled),
+                // // MSC4169: Backwards-compatible redaction sending using `/send`
+                // ("com.beeper.msc4169".to_string(), config.experimental.msc4169_enabled),
+                // // MSC4354: Sticky events
+                // ("org.matrix.msc4354".to_string(), config.experimental.msc4354_enabled),
+                // // MSC4380: Invite blocking
+                // ("org.matrix.msc4380.stable".to_string(), true),
+                // // MSC4445: Sync timeline order
+                // ("org.matrix.msc4445.initial_sync_timeline_topological_ordering".to_string(), true),
+            ]),
+        })
+    }
+}

rust/src/http_client.rs

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