Move per-homeserver Rust state into a RustRuntime object on the HomeServer

Previously the tokio runtime was stashed in a hidden attribute on the
reactor object, installed lazily by whichever Rust code first needed it,
and started via callWhenRunning (which never fires under trial's
MemoryReactorClock, hence reactor.run() workarounds in several tests).

Instead, introduce:

* `Reactor`, a typed Rust wrapper around the Twisted reactor. The duck
  type is validated once at the FFI boundary, and the wrapper is the
  single place naming the Twisted API surface Rust relies on.
* `RustRuntime`, a frozen pyclass wrapping `Arc<RustRuntimeInner>`,
  constructed once per homeserver via `hs.get_rust_runtime()`. The tokio
  runtime starts lazily on first use; shutdown is driven by a reactor
  shutdown trigger holding only a `Weak` reference (so there is no
  uncollectable reference cycle through the Rust struct), with a `Drop`
  backstop for reactors whose triggers never fire.

Rust consumers (`HttpClient`, `VersionsHandler`, the Python DB pool
wrapper) now receive the runtime or reactor handle explicitly instead of
fishing state out of reactor attributes, and the reactor.run() /
manual-startup workarounds in tests are no longer needed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HyuiV3m44ZbK9o24EnMMES

changelog.d/19969.misc

@@ -0,0 +1 @@
+Refactor how the Rust side of Synapse is tied to the reactor: per-homeserver Rust state (the tokio runtime and reactor handle) now lives in a `RustRuntime` object owned by the `HomeServer`, instead of being stashed in an attribute on the reactor.

rust/src/deferred.rs

@@ -25,7 +25,8 @@ use pyo3::{
 };
 use tokio::sync::oneshot;
 
-use crate::tokio_runtime::runtime;
+use crate::reactor::Reactor;
+use crate::runtime::RustRuntimeInner;
 
 create_exception!(
     synapse.synapse_rust.http_client,
@@ -74,7 +75,7 @@ fn logging_context_module(py: Python<'_>) -> PyResult<&Bound<'_, PyAny>> {
 /// Does not handle deferred cancellation or contextvars.
 pub fn create_deferred<'py, F, O>(
     py: Python<'py>,
-    reactor: &Bound<'py, PyAny>,
+    runtime: &Arc<RustRuntimeInner>,
     fut: F,
 ) -> PyResult<Bound<'py, PyAny>>
 where
@@ -85,12 +86,12 @@ where
     let deferred_callback = deferred.getattr("callback")?.unbind();
     let deferred_errback = deferred.getattr("errback")?.unbind();
 
-    let rt = runtime(reactor)?;
-    let handle = rt.handle()?;
+    let handle = runtime.tokio_handle()?;
     let task = handle.spawn(fut);
 
-    // Unbind the reactor so that we can pass it to the task
-    let reactor = reactor.clone().unbind();
+    // Keep the runtime state (and, through it, the reactor) alive while the
+    // task is in flight.
+    let runtime = Arc::clone(runtime);
     handle.spawn(async move {
         let res = task.await;
 
@@ -104,19 +105,18 @@ where
                 },
             };
 
-            // Re-bind the reactor
-            let reactor = reactor.bind(py);
-
             // Send the result to the deferred, via `.callback(..)` or `.errback(..)`
             match res {
                 Ok(obj) => {
-                    reactor
-                        .call_method("callFromThread", (deferred_callback, obj), None)
+                    runtime
+                        .reactor()
+                        .call_from_thread(py, (deferred_callback, obj))
                         .expect("callFromThread should not fail"); // There's nothing we can really do with errors here
                 }
                 Err(err) => {
-                    reactor
-                        .call_method("callFromThread", (deferred_errback, err), None)
+                    runtime
+                        .reactor()
+                        .call_from_thread(py, (deferred_errback, err))
                         .expect("callFromThread should not fail"); // There's nothing we can really do with errors here
                 }
             }
@@ -137,7 +137,7 @@ where
 /// 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<F>(
-    reactor: Py<PyAny>,
+    reactor: Reactor,
     make_awaitable: F,
 ) -> PyResult<Py<PyAny>>
 where
@@ -235,9 +235,7 @@ where
             },
         )?;
 
-        reactor
-            .bind(py)
-            .call_method1(intern!(py, "callFromThread"), (starter,))?;
+        reactor.call_from_thread(py, (starter,))?;
 
         Ok(())
     })?;

rust/src/handlers/mod.rs

@@ -22,6 +22,7 @@ use pyo3::{
 };
 
 use crate::config::SynapseHomeServerConfig;
+use crate::runtime::RustRuntime;
 use crate::storage::db::python_db_pool::PythonDatabasePoolWrapper;
 use crate::storage::store::Store;
 
@@ -39,16 +40,19 @@ impl RustHandlers {
     pub fn py_new(py: Python<'_>, homeserver: &Bound<'_, PyAny>) -> PyResult<RustHandlers> {
         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 = homeserver.call_method0("get_reactor")?.unbind();
+        // The per-homeserver Rust state, which gives us the tokio runtime
+        // and the Twisted reactor.
+        let runtime: Bound<'_, RustRuntime> =
+            homeserver.call_method0("get_rust_runtime")?.extract()?;
+        let runtime = Arc::clone(runtime.get().inner());
 
         // hs.get_datastores().main.db_pool
         let db_pool_py = homeserver
             .call_method0("get_datastores")?
             .getattr("main")?
             .getattr("db_pool")?;
-        let db_pool = PythonDatabasePoolWrapper::new(&db_pool_py, reactor.clone_ref(py))?;
+        let db_pool =
+            PythonDatabasePoolWrapper::new(&db_pool_py, runtime.reactor().clone_ref(py))?;
 
         // Store is shared across all of the handlers so let's use an `Arc`
         let store = Arc::new(Store {
@@ -64,7 +68,7 @@ impl RustHandlers {
             versions::VersionsHandler {
                 global_unstable_feature_map: Arc::clone(&global_unstable_feature_map),
                 store: Arc::clone(&store),
-                reactor: reactor.clone_ref(py),
+                runtime: Arc::clone(&runtime),
             },
         )?;
 

rust/src/handlers/versions.rs

@@ -21,6 +21,7 @@ use serde::Serialize;
 
 use crate::config::{types::RoomCreationPreset, SynapseHomeServerConfig};
 use crate::deferred::create_deferred;
+use crate::runtime::RustRuntimeInner;
 use crate::storage::store::{PerUserExperimentalFeature, Store};
 
 /// `GET /_matrix/client/versions` response
@@ -45,9 +46,9 @@ impl<'py> IntoPyObject<'py> for VersionsResponse {
 pub struct VersionsHandler {
     pub global_unstable_feature_map: Arc<UnstableFeatureMap>,
     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>,
+    /// The per-homeserver Rust state, used to bridge our `async` response
+    /// back into a Twisted deferred that Python can `await`.
+    pub runtime: Arc<RustRuntimeInner>,
 }
 
 #[pymethods]
@@ -63,7 +64,7 @@ impl VersionsHandler {
         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 {
+        create_deferred(py, &self.runtime, async move {
             build_versions_response(&store, &global_unstable_feature_map, user_id.as_deref())
                 .await
                 .map_err(|err| {

rust/src/http_client.rs

@@ -13,6 +13,7 @@
  */
 
 use std::collections::HashMap;
+use std::sync::Arc;
 
 use anyhow::Context;
 use http_body_util::BodyExt;
@@ -21,7 +22,7 @@ use reqwest::RequestBuilder;
 
 use crate::deferred::create_deferred;
 use crate::errors::HttpResponseException;
-use crate::tokio_runtime::runtime;
+use crate::runtime::{RustRuntime, RustRuntimeInner};
 
 /// Called when registering modules with python.
 pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
@@ -42,21 +43,18 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
 #[pyclass]
 struct HttpClient {
     client: reqwest::Client,
-    reactor: Py<PyAny>,
+    runtime: Arc<RustRuntimeInner>,
 }
 
 #[pymethods]
 impl HttpClient {
     #[new]
-    #[pyo3(signature = (reactor, user_agent, http2_only = false))]
+    #[pyo3(signature = (runtime, user_agent, http2_only = false))]
     pub fn py_new(
-        reactor: Bound<PyAny>,
+        runtime: &Bound<'_, RustRuntime>,
         user_agent: &str,
         http2_only: bool,
     ) -> PyResult<HttpClient> {
-        // Make sure the runtime gets installed
-        let _ = runtime(&reactor)?;
-
         let mut builder = reqwest::Client::builder().user_agent(user_agent);
 
         if http2_only {
@@ -69,7 +67,7 @@ impl HttpClient {
 
         Ok(HttpClient {
             client,
-            reactor: reactor.unbind(),
+            runtime: Arc::clone(runtime.get().inner()),
         })
     }
 
@@ -107,7 +105,7 @@ impl HttpClient {
         builder: RequestBuilder,
         response_limit: usize,
     ) -> PyResult<Bound<'a, PyAny>> {
-        create_deferred(py, self.reactor.bind(py), async move {
+        create_deferred(py, &self.runtime, async move {
             let response = builder.send().await.context("sending request")?;
 
             let status = response.status();

rust/src/lib.rs

@@ -19,11 +19,12 @@ pub mod json;
 pub mod matrix_const;
 pub mod msc4388_rendezvous;
 pub mod push;
+pub mod reactor;
 pub mod rendezvous;
 pub mod room_versions;
+pub mod runtime;
 pub mod segmenter;
 pub mod storage;
-pub mod tokio_runtime;
 pub mod types;
 
 lazy_static! {
@@ -76,6 +77,7 @@ fn synapse_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
     handlers::register_module(py, m)?;
     http_client::register_module(py, m)?;
     rendezvous::register_module(py, m)?;
+    runtime::register_module(py, m)?;
     msc4388_rendezvous::register_module(py, m)?;
     segmenter::register_module(py, m)?;
     room_versions::register_module(py, m)?;

rust/src/reactor.rs

@@ -0,0 +1,88 @@
+/*
+ * 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>.
+ *
+ */
+
+//! A typed wrapper around the Twisted reactor.
+
+use pyo3::{call::PyCallArgs, exceptions::PyTypeError, intern, prelude::*};
+
+/// The reactor methods Rust code relies on.
+///
+/// Extracting a [`Reactor`] from a Python object fails if any of these are
+/// missing, so mistakes surface as a `TypeError` at the FFI boundary rather
+/// than as an `AttributeError` on a tokio worker thread later.
+const REQUIRED_METHODS: &[&str] = &["callFromThread", "addSystemEventTrigger"];
+
+/// The Twisted reactor, as seen from Rust.
+///
+/// This is not a static guarantee that the object behaves like a reactor (it
+/// is a foreign Python object), but the duck type is checked once at
+/// extraction, and this module is the single place that names the Twisted
+/// API surface that Rust code depends on.
+pub struct Reactor(Py<PyAny>);
+
+impl<'a, 'py> FromPyObject<'a, 'py> for Reactor {
+    type Error = PyErr;
+
+    fn extract(obj: Borrowed<'a, 'py, PyAny>) -> PyResult<Self> {
+        for name in REQUIRED_METHODS {
+            if !obj.hasattr(*name)? {
+                return Err(PyTypeError::new_err(format!(
+                    "expected a Twisted reactor, but {} has no `{name}` method",
+                    obj.get_type()
+                )));
+            }
+        }
+
+        Ok(Reactor(obj.to_owned().unbind()))
+    }
+}
+
+impl Reactor {
+    /// `reactor.callFromThread(f, *args)`: schedule a call on the reactor
+    /// thread. This is the only reactor method that is safe to call from
+    /// other threads (e.g. tokio workers).
+    ///
+    /// `args` is the full argument tuple, starting with the callable itself.
+    pub fn call_from_thread<'py>(
+        &self,
+        py: Python<'py>,
+        args: impl PyCallArgs<'py>,
+    ) -> PyResult<()> {
+        self.0
+            .bind(py)
+            .call_method1(intern!(py, "callFromThread"), args)?;
+
+        Ok(())
+    }
+
+    /// Register `callable` to run after the reactor has shut down, via
+    /// `reactor.addSystemEventTrigger("after", "shutdown", callable)`.
+    pub fn add_shutdown_trigger(
+        &self,
+        py: Python<'_>,
+        callable: &Bound<'_, PyAny>,
+    ) -> PyResult<()> {
+        self.0.bind(py).call_method1(
+            intern!(py, "addSystemEventTrigger"),
+            (intern!(py, "after"), intern!(py, "shutdown"), callable),
+        )?;
+
+        Ok(())
+    }
+
+    pub fn clone_ref(&self, py: Python<'_>) -> Reactor {
+        Reactor(self.0.clone_ref(py))
+    }
+}

rust/src/runtime.rs

@@ -0,0 +1,191 @@
+/*
+ * 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>.
+ *
+ */
+
+//! The per-homeserver state for the Rust side of Synapse.
+//!
+//! A [`RustRuntime`] is created once per homeserver (`hs.get_rust_runtime()`)
+//! and holds everything the Rust side keeps for the lifetime of that
+//! homeserver: currently the tokio thread pool and a handle to the Twisted
+//! reactor. Rust consumers (e.g. the HTTP client) clone the inner
+//! [`Arc<RustRuntimeInner>`] at construction time and don't need the GIL (or
+//! the Python-facing object) to reach it afterwards.
+
+use std::sync::{Arc, Mutex, Weak};
+use std::time::Duration;
+
+use anyhow::Context;
+use pyo3::{exceptions::PyRuntimeError, prelude::*};
+use tokio::runtime::{Handle, Runtime};
+
+use crate::reactor::Reactor;
+
+/// How long to wait for in-flight tokio tasks when shutting down with the
+/// reactor.
+const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// State of the lazily-started tokio runtime.
+enum TokioState {
+    /// Not started yet; the runtime is built on first use.
+    NotStarted,
+    Running(Runtime),
+    /// Shut down after the reactor stopped. Cannot be restarted.
+    Shutdown,
+}
+
+/// The state shared between the Python-facing [`RustRuntime`] handle and any
+/// Rust-side consumers holding an `Arc` of this.
+pub struct RustRuntimeInner {
+    reactor: Reactor,
+    tokio: Mutex<TokioState>,
+    worker_threads: usize,
+}
+
+impl RustRuntimeInner {
+    /// The Twisted reactor this homeserver runs on.
+    pub fn reactor(&self) -> &Reactor {
+        &self.reactor
+    }
+
+    /// Get a handle to the tokio runtime, starting the runtime if it hasn't
+    /// been started yet.
+    pub fn tokio_handle(&self) -> PyResult<Handle> {
+        let mut state = self
+            .tokio
+            .lock()
+            .map_err(|_| PyRuntimeError::new_err("tokio runtime lock poisoned"))?;
+
+        match &*state {
+            TokioState::Running(runtime) => Ok(runtime.handle().clone()),
+            TokioState::NotStarted => {
+                let runtime = tokio::runtime::Builder::new_multi_thread()
+                    .worker_threads(self.worker_threads)
+                    .enable_all()
+                    .build()
+                    .context("building tokio runtime")?;
+                let handle = runtime.handle().clone();
+                *state = TokioState::Running(runtime);
+                Ok(handle)
+            }
+            TokioState::Shutdown => Err(PyRuntimeError::new_err(
+                "the tokio runtime has been shut down",
+            )),
+        }
+    }
+
+    /// Shut the tokio runtime down, waiting (with the GIL released) for
+    /// in-flight tasks to finish. Called via [`ShutdownHook`] when the
+    /// reactor shuts down.
+    fn shutdown(&self, py: Python<'_>) -> PyResult<()> {
+        let mut state = self
+            .tokio
+            .lock()
+            .map_err(|_| PyRuntimeError::new_err("tokio runtime lock poisoned"))?;
+        let previous = std::mem::replace(&mut *state, TokioState::Shutdown);
+        // Don't hold the lock while blocking on the shutdown below.
+        drop(state);
+
+        if let TokioState::Running(runtime) = previous {
+            py.detach(|| runtime.shutdown_timeout(SHUTDOWN_TIMEOUT));
+        }
+
+        Ok(())
+    }
+}
+
+impl Drop for RustRuntimeInner {
+    fn drop(&mut self) {
+        // Backstop for reactors whose shutdown trigger never fires (e.g.
+        // `MemoryReactorClock` in tests, which is never actually run).
+        // `shutdown_background` rather than a blocking shutdown, because the
+        // last `Arc` may be dropped from a task running on this very
+        // runtime, where blocking would panic.
+        if let Ok(state) = self.tokio.get_mut() {
+            if let TokioState::Running(runtime) = std::mem::replace(state, TokioState::Shutdown) {
+                runtime.shutdown_background();
+            }
+        }
+    }
+}
+
+/// The Python-facing handle to the per-homeserver Rust state.
+///
+/// Constructed by `HomeServer.get_rust_runtime()`, and passed to the Rust
+/// classes that need it (which take a clone of the inner [`Arc`] and drop
+/// this handle).
+#[pyclass(frozen, name = "RustRuntime", module = "synapse.synapse_rust")]
+pub struct RustRuntime {
+    inner: Arc<RustRuntimeInner>,
+}
+
+impl RustRuntime {
+    pub fn inner(&self) -> &Arc<RustRuntimeInner> {
+        &self.inner
+    }
+}
+
+#[pymethods]
+impl RustRuntime {
+    #[new]
+    #[pyo3(signature = (reactor, worker_threads = 4))]
+    fn py_new(py: Python<'_>, reactor: Reactor, worker_threads: usize) -> PyResult<Self> {
+        let inner = Arc::new(RustRuntimeInner {
+            reactor,
+            tokio: Mutex::new(TokioState::NotStarted),
+            worker_threads,
+        });
+
+        // Shut the tokio runtime down when the reactor does. The trigger
+        // holds only a `Weak` reference: Twisted keeping the hook alive must
+        // not keep the runtime (nor, via it, the reactor) alive, as that
+        // would be a reference cycle passing through a Rust field that
+        // Python's GC cannot see into.
+        let hook = Py::new(
+            py,
+            ShutdownHook {
+                inner: Arc::downgrade(&inner),
+            },
+        )?;
+        inner
+            .reactor
+            .add_shutdown_trigger(py, hook.bind(py).as_any())?;
+
+        Ok(RustRuntime { inner })
+    }
+}
+
+/// The callable registered with
+/// `reactor.addSystemEventTrigger("after", "shutdown", ...)`.
+#[pyclass(frozen)]
+struct ShutdownHook {
+    inner: Weak<RustRuntimeInner>,
+}
+
+#[pymethods]
+impl ShutdownHook {
+    fn __call__(&self, py: Python<'_>) -> PyResult<()> {
+        if let Some(inner) = self.inner.upgrade() {
+            inner.shutdown(py)?;
+        }
+
+        Ok(())
+    }
+}
+
+/// Called when registering modules with python.
+pub fn register_module(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_class::<RustRuntime>()?;
+
+    Ok(())
+}

rust/src/storage/db/python_db_pool.rs

@@ -40,6 +40,7 @@ use pyo3::{
 };
 
 use crate::deferred::run_python_awaitable;
+use crate::reactor::Reactor;
 use crate::storage::db::{
     DatabasePool, DbRow, DbValue, ErasedInteraction, ErasedResult, Transaction,
 };
@@ -112,13 +113,13 @@ pub struct PythonDatabasePoolWrapper {
     /// never gets garbage collected and never points back at the homeserver, so it is
     /// not part of any reference cycle. Ideally, we could worry about it but
     /// practically probably doesn't matter.
-    reactor: Py<PyAny>,
+    reactor: Reactor,
 }
 
 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: &Bound<'_, PyAny>, reactor: Py<PyAny>) -> PyResult<Self> {
+    pub fn new(database_pool: &Bound<'_, PyAny>, reactor: Reactor) -> PyResult<Self> {
         Ok(Self {
             database_pool_py_ref: PyWeakrefReference::new(database_pool)?.unbind(),
             reactor,

rust/src/tokio_runtime.rs

@@ -1,107 +0,0 @@
-/*
- * 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 anyhow::Context;
-use pyo3::prelude::*;
-use tokio::runtime::Runtime;
-
-/// This is the name of the attribute where we store the runtime on the reactor
-static TOKIO_RUNTIME_ATTR: &str = "__synapse_rust_tokio_runtime";
-
-/// A Python wrapper around a Tokio runtime.
-///
-/// This allows us to 'store' the runtime on the reactor instance, starting it
-/// when the reactor starts, and stopping it when the reactor shuts down.
-#[pyclass]
-pub struct PyTokioRuntime {
-    runtime: Option<Runtime>,
-}
-
-#[pymethods]
-impl PyTokioRuntime {
-    fn start(&mut self) -> PyResult<()> {
-        // TODO: allow customization of the runtime like the number of threads
-        let runtime = tokio::runtime::Builder::new_multi_thread()
-            .worker_threads(4)
-            .enable_all()
-            .build()?;
-
-        self.runtime = Some(runtime);
-
-        Ok(())
-    }
-
-    fn shutdown(&mut self) -> PyResult<()> {
-        let runtime = self
-            .runtime
-            .take()
-            .context("Runtime was already shutdown")?;
-
-        // Dropping the runtime will shut it down
-        drop(runtime);
-
-        Ok(())
-    }
-}
-
-impl PyTokioRuntime {
-    /// Get the handle to the Tokio runtime, if it is running.
-    pub fn handle(&self) -> PyResult<&tokio::runtime::Handle> {
-        let handle = self
-            .runtime
-            .as_ref()
-            .context("Tokio runtime is not running")?
-            .handle();
-
-        Ok(handle)
-    }
-}
-
-/// Get a handle to the Tokio runtime stored on the reactor instance, or create
-/// a new one.
-pub fn runtime<'a>(reactor: &Bound<'a, PyAny>) -> PyResult<PyRef<'a, PyTokioRuntime>> {
-    if !reactor.hasattr(TOKIO_RUNTIME_ATTR)? {
-        install_runtime(reactor)?;
-    }
-
-    get_runtime(reactor)
-}
-
-/// Install a new Tokio runtime on the reactor instance.
-fn install_runtime(reactor: &Bound<PyAny>) -> PyResult<()> {
-    let py = reactor.py();
-    let runtime = PyTokioRuntime { runtime: None };
-    let runtime = runtime.into_pyobject(py)?;
-
-    // Attach the runtime to the reactor, starting it when the reactor is
-    // running, stopping it when the reactor is shutting down
-    reactor.call_method1("callWhenRunning", (runtime.getattr("start")?,))?;
-    reactor.call_method1(
-        "addSystemEventTrigger",
-        ("after", "shutdown", runtime.getattr("shutdown")?),
-    )?;
-    reactor.setattr(TOKIO_RUNTIME_ATTR, runtime)?;
-
-    Ok(())
-}
-
-/// Get a reference to a Tokio runtime handle stored on the reactor instance.
-fn get_runtime<'a>(reactor: &Bound<'a, PyAny>) -> PyResult<PyRef<'a, PyTokioRuntime>> {
-    // This will raise if `TOKIO_RUNTIME_ATTR` is not set or if it is
-    // not a `Runtime`. Careful that this could happen if the user sets it
-    // manually, or if multiple versions of `pyo3-twisted` are used!
-    let runtime: Bound<PyTokioRuntime> = reactor.getattr(TOKIO_RUNTIME_ATTR)?.extract()?;
-    Ok(runtime.borrow())
-}

synapse/api/auth/mas.py

@@ -109,7 +109,7 @@ class MasDelegatedAuth(BaseAuth):
 
         self._http_client = hs.get_proxied_http_client()
         self._rust_http_client = HttpClient(
-            reactor=hs.get_reactor(),
+            runtime=hs.get_rust_runtime(),
             user_agent=self._http_client.user_agent.decode("utf8"),
             http2_only=self._config.force_http2,
         )

synapse/server.py

@@ -174,6 +174,7 @@ from synapse.state import StateHandler, StateResolutionHandler
 from synapse.storage import Databases
 from synapse.storage.controllers import StorageControllers
 from synapse.streams.events import EventSources
+from synapse.synapse_rust import RustRuntime
 from synapse.synapse_rust.handlers import RustHandlers
 from synapse.synapse_rust.msc4388_rendezvous import MSC4388RendezvousHandler
 from synapse.synapse_rust.rendezvous import RendezvousHandler
@@ -964,6 +965,18 @@ class HomeServer(metaclass=abc.ABCMeta):
     def get_rust_handlers(self) -> RustHandlers:
         return RustHandlers(self)
 
+    @cache_in_self
+    def get_rust_runtime(self) -> RustRuntime:
+        """The per-homeserver state for the Rust side of Synapse: the tokio
+        thread pool, plus anything else Rust code keeps for the lifetime of
+        the homeserver.
+
+        The tokio runtime is started lazily on first use, and shut down by a
+        reactor shutdown trigger.
+        """
+        # TODO: make the number of worker threads configurable
+        return RustRuntime(reactor=self.get_reactor(), worker_threads=4)
+
     @cache_in_self
     def get_event_sources(self) -> EventSources:
         return EventSources(self)

synapse/synapse_rust/__init__.pyi

@@ -1,4 +1,21 @@
+from synapse.types import ISynapseReactor
+
 def sum_as_string(a: int, b: int) -> str: ...
 def get_rust_file_digest() -> str: ...
 def reset_logging_config() -> None: ...
 def get_rustc_version() -> str: ...
+
+class RustRuntime:
+    """The per-homeserver state for the Rust side of Synapse.
+
+    Holds the tokio thread pool (started lazily on first use, shut down by a
+    reactor shutdown trigger) and a handle to the reactor. Rust classes that
+    need either take this as a constructor argument; get it from
+    `hs.get_rust_runtime()`.
+    """
+
+    def __init__(
+        self,
+        reactor: ISynapseReactor,
+        worker_threads: int = 4,
+    ) -> None: ...

synapse/synapse_rust/http_client.pyi

@@ -14,7 +14,7 @@ from typing import Mapping
 
 from twisted.internet.defer import Deferred
 
-from synapse.types import ISynapseReactor
+from synapse.synapse_rust import RustRuntime
 
 class HttpClient:
     """
@@ -23,7 +23,7 @@ class HttpClient:
 
     def __init__(
         self,
-        reactor: ISynapseReactor,
+        runtime: RustRuntime,
         user_agent: str,
         http2_only: bool = False,
     ) -> None:
@@ -31,7 +31,7 @@ class HttpClient:
         Create a new HTTP client backed by reqwest.
 
         Args:
-            reactor: The Twisted reactor to coordinate with
+            runtime: The per-homeserver Rust state (`hs.get_rust_runtime()`)
             user_agent: The user agent to use for requests
             http2_only: Whether to use HTTP/2 only, even on unencrypted connections. By
                 default, it will always use HTTP/1.1 over unencrypted connections, and

tests/handlers/test_oauth_delegation.py

@@ -216,8 +216,6 @@ class MasAuthDelegation(HomeserverTestCase):
     def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
         self.server = FakeMasServer()
         hs = self.setup_test_homeserver()
-        # This triggers the server startup hooks, which starts the Tokio thread pool
-        reactor.run()
         self._auth = checked_cast(MasDelegatedAuth, hs.get_auth())
         return hs
 

tests/push/test_http.py

@@ -1236,19 +1236,14 @@ class MSC3881VersionsTestCase(HomeserverTestCase):
     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()
+        # The tokio thread pool is started lazily on first use, so no
+        # reactor startup hooks need to run here.
         _ = HttpClient(
-            reactor=hs.get_reactor(),
+            runtime=hs.get_rust_runtime(),
             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:

tests/rest/client/test_delayed_events.py

@@ -44,19 +44,14 @@ class DelayedEventsUnstableSupportTestCase(HomeserverTestCase):
     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()
+        # The tokio thread pool is started lazily on first use, so no
+        # reactor startup hooks need to run here.
         _ = HttpClient(
-            reactor=hs.get_reactor(),
+            runtime=hs.get_rust_runtime(),
             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:

tests/rest/client/test_login_token_request.py

@@ -48,19 +48,14 @@ class LoginTokenRequestServletTestCase(unittest.HomeserverTestCase):
         self.hs.config.registration.auto_join_rooms = []
         self.hs.config.captcha.enable_registration_captcha = False
 
-        # 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 = self.hs.get_proxied_http_client()
+        # The tokio thread pool is started lazily on first use, so no
+        # reactor startup hooks need to run here.
         _ = HttpClient(
-            reactor=self.hs.get_reactor(),
+            runtime=self.hs.get_rust_runtime(),
             user_agent=self._http_client.user_agent.decode("utf8"),
         )
 
-        # This triggers the server startup hooks, which starts the Tokio thread pool
-        reactor.run()
-
         return self.hs
 
     def tearDown(self) -> None:

tests/rest/client/test_matrixrtc.py

@@ -115,19 +115,14 @@ class MatrixRtcVersionsTestCase(HomeserverTestCase):
     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()
+        # The tokio thread pool is started lazily on first use, so no
+        # reactor startup hooks need to run here.
         _ = HttpClient(
-            reactor=hs.get_reactor(),
+            runtime=hs.get_rust_runtime(),
             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:

tests/rest/client/test_versions.py

@@ -40,19 +40,14 @@ class VersionsTestCase(unittest.HomeserverTestCase):
     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()
+        # The tokio thread pool is started lazily on first use, so no
+        # reactor startup hooks need to run here.
         _ = HttpClient(
-            reactor=hs.get_reactor(),
+            runtime=hs.get_rust_runtime(),
             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:

tests/synapse_rust/test_http_client.py

@@ -89,19 +89,14 @@ class HttpClientTestCase(HomeserverTestCase):
     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()
+        # The tokio thread pool is started lazily on first use, so no
+        # reactor startup hooks need to run here.
         self._rust_http_client = HttpClient(
-            reactor=hs.get_reactor(),
+            runtime=hs.get_rust_runtime(),
             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 prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: