diff --git a/Cargo.lock b/Cargo.lock index 07f71075bd9..87f50ea5dfa 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/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. 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/config/mod.rs b/rust/src/config/mod.rs new file mode 100644 index 00000000000..62370571279 --- /dev/null +++ b/rust/src/config/mod.rs @@ -0,0 +1,75 @@ +/* + * 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 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: +/// ```no_run +/// let config: SynapseHomeServerConfig = homeserver.getattr("config")?.extract()?; +/// ``` +#[derive(FromPyObject, Clone)] +pub struct SynapseHomeServerConfig { + pub room: RoomConfig, + pub auth: AuthConfig, + pub server: ServerConfig, + pub experimental: ExperimentalConfig, +} + +#[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, + pub msc4491_enabled: bool, + pub msc4143_enabled: bool, +} diff --git a/rust/src/config/types.rs b/rust/src/config/types.rs new file mode 100644 index 00000000000..8990b52f656 --- /dev/null +++ b/rust/src/config/types.rs @@ -0,0 +1,60 @@ +/* + * 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::PyAssertionError, 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, + 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(PyAssertionError::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() + } +} diff --git a/rust/src/deferred.rs b/rust/src/deferred.rs index 91d5a08f7db..62a1ce6b903 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,6 +58,16 @@ 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. /// @@ -110,18 +127,161 @@ 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() + .map_err(|err| { + anyhow::anyhow!("Failed to acquire lock on `success_sender`: {:#}", err) + })? + .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() + .map_err(|err| { + anyhow::anyhow!("Failed to acquire lock on `error_sender`: {:#}", err) + })? + .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 +293,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(()) } 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..db8bca82d68 --- /dev/null +++ b/rust/src/handlers/versions.rs @@ -0,0 +1,335 @@ +/* + * 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::Serialize; + +use crate::config::{types::RoomCreationPreset, SynapseHomeServerConfig}; +use crate::deferred::create_deferred; +use crate::storage::store::{PerUserExperimentalFeature, Store}; + +/// `GET /_matrix/client/versions` response +#[derive(Serialize, Clone, Debug)] +struct VersionsResponse { + versions: Vec, + /// as per MSC1497 + unstable_features: UnstableFeatureMap, +} + +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, + ..*global_unstable_feature_map + }; + + 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: 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, + /// MSC4491: Invite reasons in room creation + #[serde(rename = "uk.timedout.msc4491.create_room_invite_reasons")] + msc4491_enabled: bool, + /// MSC4143: Matrix RTC transports (LiveKit backend) + #[serde(rename = "org.matrix.msc4143")] + msc4143_enabled: 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, + msc4491_enabled: config.experimental.msc4491_enabled, + msc4143_enabled: config.experimental.msc4143_enabled, + 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 eb2d8cecdde..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; @@ -20,6 +22,7 @@ pub mod push; pub mod rendezvous; pub mod room_versions; pub mod segmenter; +pub mod storage; pub mod tokio_runtime; pub mod types; @@ -70,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/rust/src/storage/db/mod.rs b/rust/src/storage/db/mod.rs new file mode 100644 index 00000000000..69d9d155a32 --- /dev/null +++ b/rust/src/storage/db/mod.rs @@ -0,0 +1,256 @@ +/* + * 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; + +pub mod python_db_pool; + +/// 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: + /// ```no_run + /// 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?; + boxed.downcast::().map(|b| *b).map_err(|_| { + anyhow::anyhow!( + "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) => 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"), + } + } +} + +impl FromDbValue for i64 { + fn from_value(value: DbValue) -> Result { + match value { + DbValue::Int(i) => Ok(i), + 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), + 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/db/python_db_pool.rs b/rust/src/storage/db/python_db_pool.rs new file mode 100644 index 00000000000..5e62b066561 --- /dev/null +++ b/rust/src/storage/db/python_db_pool.rs @@ -0,0 +1,371 @@ +/* + * 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::sync::{Arc, Mutex}; + +use anyhow::Context; +use futures::FutureExt; +use once_cell::sync::OnceCell; +use pyo3::{ + exceptions::{PyAssertionError, 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, +}; + +/// 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 { + Sqlite, + Postgres, +} + +impl DatabaseEngine { + 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 func(&mut txn).now_or_never() { + 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)) => { + // 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); + 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 => { + 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 \ + 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() + .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 + // 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)), + } + } +} + +/// 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 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. +pub struct LoggingTransactionWrapper { + /// The underlying `LoggingTransaction` + /// + /// We purposely avoid `Bound<'py, PyAny>` so it can be stored and moved freely + /// across threads (as required by `Transaction` trait). + 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 { + /// Calls the Python `LoggingTransaction.execute` function. + 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()? + ))) + } +} diff --git a/rust/src/storage/mod.rs b/rust/src/storage/mod.rs new file mode 100644 index 00000000000..735fcecb266 --- /dev/null +++ b/rust/src/storage/mod.rs @@ -0,0 +1,17 @@ +/* + * 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; +pub mod store; diff --git a/rust/src/storage/store.rs b/rust/src/storage/store.rs new file mode 100644 index 00000000000..b339d7748c5 --- /dev/null +++ b/rust/src/storage/store.rs @@ -0,0 +1,110 @@ +/* + * 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> { + // 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( + r#" + 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)?), + 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(), + ); + } + }; + + Ok(enabled) + } + .boxed() + }) + .await?; + + Ok(is_feature_enabled_for_user) + } +} 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 809e920a2bb..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,117 +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), - # MSC4143: Matrix RTC transports (LiveKit backend) - "org.matrix.msc4143": self.config.experimental.msc4143_enabled, - # 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, - "uk.timedout.msc4491.create_room_invite_reasons": self.config.experimental.msc4491_enabled, - }, - }, + versions_response_body, ) diff --git a/synapse/server.py b/synapse/server.py index bd4337c6e69..b756223e54a 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 @@ -959,6 +960,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..d6560984690 --- /dev/null +++ b/tests/rest/client/test_versions.py @@ -0,0 +1,174 @@ +# 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", + ) + 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", + 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", + 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", + 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", + 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 ab4aae02185..92283eecffb 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,85 @@ 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 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` """ - end_time = self._reactor.seconds() + timeout_ms / 1000.0 + 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 self._reactor.seconds() > end_time: + if ( + # Exceeded the Twisted reactor time timeout + # + # 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 time.time() > start_real_time_seconds + real_time_timeout.as_secs() + ): 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. + # + # 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 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 + # 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) + + loop_count += 1 def extract_cookies(self, cookies: MutableMapping[str, str]) -> None: """Process the contents of any Set-Cookie headers in the response