diff --git a/changelog.d/rust-sss-prereqs.misc b/changelog.d/rust-sss-prereqs.misc new file mode 100644 index 0000000000..6e61edd5fb --- /dev/null +++ b/changelog.d/rust-sss-prereqs.misc @@ -0,0 +1 @@ +Add the prerequisites for computing simplified sliding sync responses in Rust, behind `experimental_features.rust_sliding_sync`. diff --git a/docker/complement/conf/start_for_complement.sh b/docker/complement/conf/start_for_complement.sh index db8d274675..e6ef29595b 100755 --- a/docker/complement/conf/start_for_complement.sh +++ b/docker/complement/conf/start_for_complement.sh @@ -6,7 +6,7 @@ set -e echo "Complement Synapse launcher" echo " Args: $*" -echo " Env: SYNAPSE_COMPLEMENT_DATABASE=$SYNAPSE_COMPLEMENT_DATABASE SYNAPSE_COMPLEMENT_USE_WORKERS=$SYNAPSE_COMPLEMENT_USE_WORKERS SYNAPSE_COMPLEMENT_USE_ASYNCIO_REACTOR=$SYNAPSE_COMPLEMENT_USE_ASYNCIO_REACTOR SYNAPSE_USE_RUST_DRIVER=$SYNAPSE_USE_RUST_DRIVER SYNAPSE_RUST_STATE_RES=$SYNAPSE_RUST_STATE_RES SYNAPSE_RUST_SYNC=$SYNAPSE_RUST_SYNC" +echo " Env: SYNAPSE_COMPLEMENT_DATABASE=$SYNAPSE_COMPLEMENT_DATABASE SYNAPSE_COMPLEMENT_USE_WORKERS=$SYNAPSE_COMPLEMENT_USE_WORKERS SYNAPSE_COMPLEMENT_USE_ASYNCIO_REACTOR=$SYNAPSE_COMPLEMENT_USE_ASYNCIO_REACTOR SYNAPSE_USE_RUST_DRIVER=$SYNAPSE_USE_RUST_DRIVER SYNAPSE_RUST_STATE_RES=$SYNAPSE_RUST_STATE_RES SYNAPSE_RUST_SYNC=$SYNAPSE_RUST_SYNC SYNAPSE_RUST_SLIDING_SYNC=$SYNAPSE_RUST_SLIDING_SYNC" function log { d=$(printf '%(%Y-%m-%d %H:%M:%S)T,%.3s\n' ${EPOCHREALTIME/./ }) @@ -141,6 +141,11 @@ if [[ -n "$SYNAPSE_RUST_SYNC" ]]; then sed -i "s/^experimental_features:$/experimental_features:\n rust_sync: \"$SYNAPSE_RUST_SYNC\"/" /conf/shared.yaml.j2 fi +# Compute sliding sync responses in Rust (`experimental_features.rust_sliding_sync`). +if [[ -n "$SYNAPSE_RUST_SLIDING_SYNC" ]]; then + sed -i "s/^experimental_features:$/experimental_features:\n rust_sliding_sync: \"$SYNAPSE_RUST_SLIDING_SYNC\"/" /conf/shared.yaml.j2 +fi + # Add a directory for tests to add config overrides if they want mkdir --parents /conf/homeserver.d export _SYNAPSE_COMPLEMENT_EXTRA_CONFIG_DIR=/conf/homeserver.d diff --git a/rust/src/handlers/mod.rs b/rust/src/handlers/mod.rs index 4176372020..64a034109f 100644 --- a/rust/src/handlers/mod.rs +++ b/rust/src/handlers/mod.rs @@ -23,6 +23,7 @@ use pyo3::{ }; use crate::homeserver::HomeServer; +use crate::sliding_sync::SlidingSyncHandler; use crate::state::StateResolutionHandler; use crate::sync::SyncHandler; @@ -35,6 +36,9 @@ struct RustHandlers { state_resolution: Option>, /// Only when the main database uses the native Rust driver. sync: Option>, + /// Only when the main database uses the native Rust driver. Shares + /// `sync`'s store. + sliding_sync: Option>, } #[pymethods] @@ -74,31 +78,40 @@ impl RustHandlers { let rust_connection_pool_class = py .import("synapse.storage.rust_pool")? .getattr("RustConnectionPool")?; - let (state_resolution, sync) = - if connection_pool.is_instance(&rust_connection_pool_class)? { - let connection_pool = connection_pool.unbind(); - ( - Some(Py::new( - py, - StateResolutionHandler::new( - connection_pool.clone_ref(py), - runtime.clone(), - Arc::clone(&store), - ), - )?), - Some(Py::new( - py, - SyncHandler::new(connection_pool, runtime.clone(), store, &config), - )?), - ) - } else { - (None, None) - }; + let (state_resolution, sync, sliding_sync) = if connection_pool + .is_instance(&rust_connection_pool_class)? + { + let connection_pool = connection_pool.unbind(); + let state_resolution = Py::new( + py, + StateResolutionHandler::new( + connection_pool.clone_ref(py), + runtime.clone(), + Arc::clone(&store), + ), + )?; + let sync = SyncHandler::new( + connection_pool.clone_ref(py), + runtime.clone(), + store, + &config, + ); + let sliding_sync = + SlidingSyncHandler::new(connection_pool, runtime.clone(), Arc::clone(sync.store())); + ( + Some(state_resolution), + Some(Py::new(py, sync)?), + Some(Py::new(py, sliding_sync)?), + ) + } else { + (None, None, None) + }; Ok(RustHandlers { versions, state_resolution, sync, + sliding_sync, }) } @@ -120,6 +133,13 @@ impl RustHandlers { fn sync(&self, py: Python<'_>) -> Option> { self.sync.as_ref().map(|h| h.clone_ref(py)) } + + /// The Rust sliding sync handler, or `None` if the main database does + /// not use the native Rust driver. + #[getter] + fn sliding_sync(&self, py: Python<'_>) -> Option> { + self.sliding_sync.as_ref().map(|h| h.clone_ref(py)) + } } /// Called when registering modules with python. diff --git a/rust/src/lib.rs b/rust/src/lib.rs index f65453640c..006936c7e0 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -35,6 +35,7 @@ pub mod rendezvous; pub mod room_versions; pub mod runtime; pub mod segmenter; +pub mod sliding_sync; pub mod state; pub mod storage; pub mod sync; @@ -109,6 +110,7 @@ fn synapse_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { storage::python::register_module(py, m)?; msc4388_rendezvous::register_module(py, m)?; segmenter::register_module(py, m)?; + sliding_sync::register_module(py, m)?; state::register_module(py, m)?; sync::register_module(py, m)?; room_versions::register_module(py, m)?; diff --git a/rust/src/sliding_sync/mod.rs b/rust/src/sliding_sync/mod.rs new file mode 100644 index 0000000000..024890bf1e --- /dev/null +++ b/rust/src/sliding_sync/mod.rs @@ -0,0 +1,196 @@ +/* + * 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: + * . + */ + +//! Simplified sliding sync (MSC4186) in Rust, reading the database through +//! the native Postgres driver. +//! +//! [`SlidingSyncHandler`] is the Python entry point, used by +//! `synapse.handlers.sliding_sync.SlidingSyncHandler` when +//! `rust_sliding_sync` is on. It shares the v2 sync handler's +//! [`SyncStore`], so the two read through the same caches. +//! +//! The handler's entry points follow the v2 handler's shape: a `start_*` +//! method reads what it needs from Python and returns a future, which the +//! public method runs off the GIL on the runtime and returns as a deferred +//! ([`SlidingSyncHandler::defer`]), and a `_blocking` twin runs to completion +//! for tests ([`SlidingSyncHandler::block_on`]). + +use std::sync::Arc; + +use futures::future::BoxFuture; +use pyo3::{exceptions::PyRuntimeError, intern, prelude::*, types::PyModule}; + +use crate::database::postgres::pool::{ConnectionPool, PyConnectionPool}; +use crate::deferred::{create_deferred_with, OnCancel}; +use crate::runtime::RustRuntime; +use crate::storage::sync_store::{SyncDbStats, SyncStore}; + +/// Computes sliding sync responses with the Rust implementation, reading +/// the database through the native Postgres driver. +#[pyclass(frozen)] +pub struct SlidingSyncHandler { + /// The Python `RustConnectionPool`. Its `_pool` is read per call, as + /// `close()`/`start()` replace it. + db_pool: Py, + runtime: RustRuntime, + /// The queries, over the Rust store's caches. The v2 sync handler's. + store: Arc, +} + +impl SlidingSyncHandler { + pub fn new(db_pool: Py, runtime: RustRuntime, store: Arc) -> Self { + SlidingSyncHandler { + db_pool, + runtime, + store, + } + } + + pub fn store(&self) -> &Arc { + &self.store + } + + /// The connection pool to read through, or an error if it is closed. + pub fn pool(&self, py: Python<'_>) -> PyResult { + let pool = self.db_pool.bind(py).getattr(intern!(py, "_pool"))?; + if pool.is_none() { + return Err(PyRuntimeError::new_err("the database pool is closed")); + } + Ok(pool.cast::()?.get().pool.clone()) + } + + /// Run `future` on the runtime and return a deferred for its result. + /// Cancelling the deferred stops the computation. The servlet is not + /// `@cancellable`, so today that only happens on shutdown. + pub fn defer<'py, O>( + &self, + py: Python<'py>, + future: BoxFuture<'static, PyResult>, + ) -> PyResult> + where + for<'a> O: IntoPyObject<'a> + Send + 'static, + { + create_deferred_with(py, &self.runtime, future, OnCancel::Abort) + } + + /// Run `future` to completion on the runtime, releasing the GIL. For the + /// `_blocking` entry points. + pub fn block_on( + &self, + py: Python<'_>, + future: BoxFuture<'static, PyResult>, + ) -> PyResult { + let handle = self.runtime.tokio_handle()?; + py.detach(|| handle.block_on(future)) + } +} + +/// What the Rust implementation did for one request, for the +/// `synapse_sliding_sync_rust_*` histograms. Python observes it once per +/// request. +#[pyclass(frozen, get_all, skip_from_py_object)] +#[derive(Debug, Default, Clone, Copy, PartialEq)] +pub struct SlidingSyncStats { + /// Time spent waiting on statements, in seconds. + pub db_time: f64, + /// Time spent waiting for a connection from the pool, in seconds. + pub checkout_time: f64, + /// Transactions begun. + pub transactions: u64, + /// Statements executed. + pub queries: u64, + /// Rooms whose entry Rust computed. + pub rooms_computed: u64, + /// Rooms whose lazy-loaded members were looked up in the connection + /// state. + pub lazy_member_lookups: u64, +} + +impl SlidingSyncStats { + /// The database half of the stats, from a reader's. + pub fn from_db(db: &SyncDbStats) -> Self { + SlidingSyncStats { + db_time: db.db_time.as_secs_f64(), + checkout_time: db.checkout_time.as_secs_f64(), + transactions: db.transactions, + queries: db.queries, + ..Default::default() + } + } +} + +#[pymethods] +impl SlidingSyncStats { + #[new] + #[pyo3(signature = (*, db_time=0.0, checkout_time=0.0, transactions=0, queries=0, rooms_computed=0, lazy_member_lookups=0))] + fn py_new( + db_time: f64, + checkout_time: f64, + transactions: u64, + queries: u64, + rooms_computed: u64, + lazy_member_lookups: u64, + ) -> Self { + SlidingSyncStats { + db_time, + checkout_time, + transactions, + queries, + rooms_computed, + lazy_member_lookups, + } + } +} + +/// Called when registering modules with python. +pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { + let child_module = PyModule::new(py, "sliding_sync")?; + child_module.add_class::()?; + 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 sliding_sync` work. + py.import("sys")? + .getattr("modules")? + .set_item("synapse.synapse_rust.sliding_sync", child_module)?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + + #[test] + fn stats_from_db() { + let db = SyncDbStats { + checkouts: 3, + checkout_time: Duration::from_millis(5), + transactions: 2, + queries: 7, + db_time: Duration::from_millis(250), + ..Default::default() + }; + let stats = SlidingSyncStats::from_db(&db); + assert_eq!(stats.transactions, 2); + assert_eq!(stats.queries, 7); + assert!((stats.db_time - 0.25).abs() < 1e-9); + assert!((stats.checkout_time - 0.005).abs() < 1e-9); + assert_eq!(stats.rooms_computed, 0); + } +} diff --git a/rust/src/storage/sync_store.rs b/rust/src/storage/sync_store.rs index bd7836cd83..b06cf70a87 100644 --- a/rust/src/storage/sync_store.rs +++ b/rust/src/storage/sync_store.rs @@ -196,6 +196,8 @@ pub struct SyncDbStats { pub checkouts: u64, /// Time spent waiting for a connection. pub checkout_time: Duration, + /// Transactions begun. + pub transactions: u64, /// Statements executed. pub queries: u64, /// Rows returned. @@ -273,6 +275,7 @@ impl<'a> SyncReader<'a> { .execute("BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY") .await?; borrowed.in_transaction = true; + self.update_stats(|stats| stats.transactions += 1); Ok(borrowed) } diff --git a/scripts-dev/complement.sh b/scripts-dev/complement.sh index a13f480f22..d5c32281cd 100755 --- a/scripts-dev/complement.sh +++ b/scripts-dev/complement.sh @@ -376,6 +376,12 @@ main() { export PASS_SYNAPSE_RUST_SYNC="$RUST_SYNC" fi + if [[ -n "$RUST_SLIDING_SYNC" ]]; then + # Set `experimental_features.rust_sliding_sync` (off or on). The Rust + # sliding sync handler needs RUST_DRIVER=1. + export PASS_SYNAPSE_RUST_SLIDING_SYNC="$RUST_SLIDING_SYNC" + fi + if [[ -n "$UNIX_SOCKETS" ]]; then # Enable full on Unix socket mode for Synapse, Redis and Postgresql export PASS_SYNAPSE_USE_UNIX_SOCKET=1 diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index 1410e6a5f4..7df16c75ce 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -369,6 +369,24 @@ class ExperimentalConfig(Config): ) self.rust_sync: Literal["off", "shadow", "on"] = rust_sync + # Whether to compute simplified sliding sync (MSC4186) responses in + # Rust. Needs the native Rust Postgres driver. + # - "off": the Python implementation. + # - "on": Rust, falling back to Python per room or per request where + # Rust cannot compute it. + # There is no shadow mode. + rust_sliding_sync = experimental.get("rust_sliding_sync", "off") + if rust_sliding_sync is True: + rust_sliding_sync = "on" + elif rust_sliding_sync is False: + rust_sliding_sync = "off" + if rust_sliding_sync not in ("off", "on"): + raise ConfigError( + "experimental_features.rust_sliding_sync must be 'off' or 'on'", + ("experimental", "rust_sliding_sync"), + ) + self.rust_sliding_sync: Literal["off", "on"] = rust_sliding_sync + # How much of outbound federation runs in Rust. One ordered level, each # implying the ones before it: # - "off": all Python. diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index afb93e42b4..59b88d90de 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -18,7 +18,7 @@ from itertools import chain from typing import TYPE_CHECKING, AbstractSet, Mapping import attr -from prometheus_client import Histogram +from prometheus_client import Counter, Histogram from typing_extensions import assert_never from synapse.api.constants import Direction, EventTypes, Membership @@ -46,6 +46,10 @@ from synapse.storage.databases.main.stream import PaginateFunction from synapse.storage.roommember import ( MemberSummary, ) +from synapse.synapse_rust.sliding_sync import ( + SlidingSyncHandler as RustSlidingSyncHandler, + SlidingSyncStats as RustSlidingSyncStats, +) from synapse.types import ( JsonDict, MutableStateMap, @@ -86,6 +90,82 @@ sync_processing_time = Histogram( labelnames=["initial", SERVER_NAME_LABEL], ) +# Room entries computed in `rust_sliding_sync: on` mode. `result` is "rust", +# "fallback" for a room left to Python, or "error" for a room Python computed +# after the Rust handler failed on it. +rust_sliding_sync_rooms_counter = Counter( + "synapse_sliding_sync_rust_rooms_total", + "Room entries of sliding sync responses computed in `rust_sliding_sync: on` " + "mode, by whether Rust computed them or fell back to Python", + labelnames=["result", SERVER_NAME_LABEL], +) + +# Requests served in `rust_sliding_sync: on` mode. `result` is "rust" for a +# request Rust took part in, "fallback" for one Python computed in full because +# Rust does not cover it (the sliding sync background updates have not finished, +# or a list uses a filter Python does not implement either), or "error" for one +# Python computed in full after the Rust handler failed on it. +rust_sliding_sync_requests_counter = Counter( + "synapse_sliding_sync_rust_requests_total", + "Sliding sync requests in `rust_sliding_sync: on` mode, by whether Rust " + "took part or Python computed the whole response", + labelnames=["result", SERVER_NAME_LABEL], +) + +# What the Rust implementation did per request, from its `SlidingSyncStats`. +rust_sliding_sync_db_time = Histogram( + "synapse_sliding_sync_rust_db_seconds", + "Time a Rust sliding sync request spent waiting on database statements", + labelnames=[SERVER_NAME_LABEL], +) +rust_sliding_sync_checkout_time = Histogram( + "synapse_sliding_sync_rust_db_checkout_seconds", + "Time a Rust sliding sync request spent waiting for database connections", + labelnames=[SERVER_NAME_LABEL], +) +_RUST_COUNT_BUCKETS = (0, 1, 2, 5, 10, 20, 50, 100, 200, 500, 1000) +rust_sliding_sync_transactions = Histogram( + "synapse_sliding_sync_rust_db_transactions", + "Database transactions begun per Rust sliding sync request", + labelnames=[SERVER_NAME_LABEL], + buckets=_RUST_COUNT_BUCKETS, +) +rust_sliding_sync_queries = Histogram( + "synapse_sliding_sync_rust_db_queries", + "Database statements executed per Rust sliding sync request", + labelnames=[SERVER_NAME_LABEL], + buckets=_RUST_COUNT_BUCKETS, +) +rust_sliding_sync_rooms_computed = Histogram( + "synapse_sliding_sync_rust_rooms_computed", + "Room entries the Rust implementation computed per sliding sync request", + labelnames=[SERVER_NAME_LABEL], + buckets=_RUST_COUNT_BUCKETS, +) +rust_sliding_sync_lazy_member_lookups = Histogram( + "synapse_sliding_sync_rust_lazy_member_lookups", + "Rooms whose lazy-loaded members the Rust implementation looked up in the " + "connection state, per sliding sync request", + labelnames=[SERVER_NAME_LABEL], + buckets=_RUST_COUNT_BUCKETS, +) + + +def observe_rust_sliding_sync_stats( + stats: RustSlidingSyncStats, server_name: str +) -> None: + """Record one request's `SlidingSyncStats` in the histograms.""" + labels = {SERVER_NAME_LABEL: server_name} + rust_sliding_sync_db_time.labels(**labels).observe(stats.db_time) + rust_sliding_sync_checkout_time.labels(**labels).observe(stats.checkout_time) + rust_sliding_sync_transactions.labels(**labels).observe(stats.transactions) + rust_sliding_sync_queries.labels(**labels).observe(stats.queries) + rust_sliding_sync_rooms_computed.labels(**labels).observe(stats.rooms_computed) + rust_sliding_sync_lazy_member_lookups.labels(**labels).observe( + stats.lazy_member_lookups + ) + + # Limit the number of state_keys we should remember sending down the connection for each # (room_id, user_id). We don't want to store and pull out too much data in the database. # @@ -113,6 +193,23 @@ class SlidingSyncHandler: self.extensions = SlidingSyncExtensionHandler(hs) self.room_lists = SlidingSyncRoomLists(hs) + # The Rust sliding sync handler, used when the `rust_sliding_sync` + # experimental option is "on". It needs the native Rust Postgres driver. + self._rust_sliding_sync_mode = hs.config.experimental.rust_sliding_sync + self._rust_sliding_sync: RustSlidingSyncHandler | None = None + if self._rust_sliding_sync_mode != "off": + self._rust_sliding_sync = hs.get_rust_handlers().sliding_sync + if self._rust_sliding_sync is None: + logger.warning( + "rust_sliding_sync is %r but the main database does not use " + "the native Rust driver: using the Python sliding sync handler", + self._rust_sliding_sync_mode, + ) + + # Whether a Rust error raises rather than falling back to Python. Tests + # set it. + self._rust_sliding_sync_strict = False + async def wait_for_sync_for_user( self, requester: Requester, diff --git a/synapse/synapse_rust/handlers.pyi b/synapse/synapse_rust/handlers.pyi index 29531b2dd7..eb39802217 100644 --- a/synapse/synapse_rust/handlers.pyi +++ b/synapse/synapse_rust/handlers.pyi @@ -14,6 +14,7 @@ from typing import TYPE_CHECKING, Optional from twisted.internet.defer import Deferred +from synapse.synapse_rust.sliding_sync import SlidingSyncHandler from synapse.synapse_rust.state import StateResolutionHandler from synapse.synapse_rust.sync import SyncHandler from synapse.types import JsonDict @@ -44,3 +45,8 @@ class RustHandlers: def sync(self) -> SyncHandler | None: """The Rust sync handler, or None if the main database does not use the native Rust Postgres driver.""" + + @property + def sliding_sync(self) -> SlidingSyncHandler | None: + """The Rust sliding sync handler, or None if the main database does not + use the native Rust Postgres driver.""" diff --git a/synapse/synapse_rust/sliding_sync.pyi b/synapse/synapse_rust/sliding_sync.pyi new file mode 100644 index 0000000000..138c72e6e9 --- /dev/null +++ b/synapse/synapse_rust/sliding_sync.pyi @@ -0,0 +1,45 @@ +# 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: +# . + +"""Simplified sliding sync (MSC4186) in Rust. See `rust/src/sliding_sync/`.""" + +class SlidingSyncHandler: + """Computes sliding sync responses in Rust, reading the database through + the native Postgres driver. Built by `RustHandlers`.""" + +class SlidingSyncStats: + """What the Rust implementation did for one request.""" + + def __init__( + self, + *, + db_time: float = 0.0, + checkout_time: float = 0.0, + transactions: int = 0, + queries: int = 0, + rooms_computed: int = 0, + lazy_member_lookups: int = 0, + ) -> None: ... + @property + def db_time(self) -> float: + """Time spent waiting on statements, in seconds.""" + @property + def checkout_time(self) -> float: + """Time spent waiting for a database connection, in seconds.""" + @property + def transactions(self) -> int: ... + @property + def queries(self) -> int: ... + @property + def rooms_computed(self) -> int: ... + @property + def lazy_member_lookups(self) -> int: ... diff --git a/tests/config/test_experimental.py b/tests/config/test_experimental.py index b6188859ee..c9a4045ef8 100644 --- a/tests/config/test_experimental.py +++ b/tests/config/test_experimental.py @@ -74,6 +74,25 @@ class ExperimentalConfigTestCase(unittest.TestCase): with self.assertRaises(ConfigError): _read_config({"experimental_features": {"rust_state_res": "maybe"}}) + @parameterized.expand( + [ + # YAML reads these as booleans. + ("rust_sliding_sync: on", "on"), + ("rust_sliding_sync: off", "off"), + ("rust_sliding_sync: 'on'", "on"), + ("{}", "off"), + ] + ) + def test_rust_sliding_sync(self, features_yaml: str, expected: str) -> None: + config = _read_config({"experimental_features": yaml.safe_load(features_yaml)}) + self.assertEqual(config.rust_sliding_sync, expected) + + @parameterized.expand([("shadow",), ("maybe",)]) + def test_rust_sliding_sync_rejects_unknown_modes(self, mode: str) -> None: + """There is no shadow mode.""" + with self.assertRaises(ConfigError): + _read_config({"experimental_features": {"rust_sliding_sync": mode}}) + def _read_config(config_values: JsonDict) -> ExperimentalConfig: config = ExperimentalConfig(RootConfig()) diff --git a/tests/handlers/test_sliding_sync_rust.py b/tests/handlers/test_sliding_sync_rust.py new file mode 100644 index 0000000000..20bfe04004 --- /dev/null +++ b/tests/handlers/test_sliding_sync_rust.py @@ -0,0 +1,79 @@ +# +# 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: +# . +# + +"""The wiring of the Rust sliding sync handler (`rust/src/sliding_sync/`).""" + +from unittest import SkipTest + +from prometheus_client import REGISTRY + +from synapse.handlers.sliding_sync import observe_rust_sliding_sync_stats +from synapse.synapse_rust.sliding_sync import SlidingSyncStats + +import tests.utils +from tests import unittest +from tests.unittest import override_config + + +class RustSlidingSyncHandlerTestCase(unittest.HomeserverTestCase): + """The Rust sliding sync handler is only wired up on the native Rust + driver.""" + + @override_config({"experimental_features": {"rust_sliding_sync": "on"}}) + def test_needs_the_rust_driver(self) -> None: + if tests.utils.RUST_SLIDING_SYNC_FOR_TESTS: + raise SkipTest("SYNAPSE_TEST_RUST_SLIDING_SYNC overrides the config") + handler = self.hs.get_sliding_sync_handler() + self.assertEqual(handler._rust_sliding_sync_mode, "on") + if tests.utils.USE_RUST_DRIVER_FOR_TESTS: + self.assertIsNotNone(handler._rust_sliding_sync) + self.assertIsNotNone(self.hs.get_rust_handlers().sliding_sync) + else: + # Falls back to the Python handler, with a warning. + self.assertIsNone(handler._rust_sliding_sync) + self.assertIsNone(self.hs.get_rust_handlers().sliding_sync) + + def test_off_by_default(self) -> None: + handler = self.hs.get_sliding_sync_handler() + # Unless the test run sets it for every homeserver. + expected = tests.utils.RUST_SLIDING_SYNC_FOR_TESTS or "off" + self.assertEqual(handler._rust_sliding_sync_mode, expected) + if expected == "off": + self.assertIsNone(handler._rust_sliding_sync) + + def test_observe_stats(self) -> None: + def sample(name: str) -> float: + value = REGISTRY.get_sample_value(name, {"server_name": "stats.test"}) + return value or 0.0 + + before_db = sample("synapse_sliding_sync_rust_db_seconds_sum") + before_queries = sample("synapse_sliding_sync_rust_db_queries_sum") + before_rooms = sample("synapse_sliding_sync_rust_rooms_computed_count") + + stats = SlidingSyncStats( + db_time=0.5, queries=12, transactions=2, rooms_computed=3 + ) + self.assertEqual(stats.queries, 12) + self.assertEqual(stats.lazy_member_lookups, 0) + observe_rust_sliding_sync_stats(stats, "stats.test") + + self.assertAlmostEqual( + sample("synapse_sliding_sync_rust_db_seconds_sum") - before_db, 0.5 + ) + self.assertEqual( + sample("synapse_sliding_sync_rust_db_queries_sum") - before_queries, 12 + ) + self.assertEqual( + sample("synapse_sliding_sync_rust_rooms_computed_count") - before_rooms, 1 + ) diff --git a/tests/server.py b/tests/server.py index e99c2cc195..2b330918e3 100644 --- a/tests/server.py +++ b/tests/server.py @@ -107,6 +107,9 @@ from synapse.storage.database import ( from synapse.storage.engines import BaseDatabaseEngine, create_engine from synapse.storage.prepare_database import prepare_database from synapse.synapse_rust.runtime import RustRuntime, set_virtual_time_msec +from synapse.synapse_rust.sliding_sync import ( + SlidingSyncHandler as RustSlidingSyncHandler, +) from synapse.synapse_rust.state import ( StateResolutionHandler as RustStateResolutionHandler, ) @@ -124,6 +127,8 @@ from tests.utils import ( POSTGRES_PASSWORD, POSTGRES_PORT, POSTGRES_USER, + RUST_SLIDING_SYNC_FOR_TESTS, + RUST_SLIDING_SYNC_STRICT_FOR_TESTS, RUST_STATE_RES_FOR_TESTS, RUST_SYNC_FOR_TESTS, RUST_SYNC_STRICT_FOR_TESTS, @@ -1328,6 +1333,13 @@ def setup_test_homeserver( Literal["off", "shadow", "on"], RUST_SYNC_FOR_TESTS ) + if RUST_SLIDING_SYNC_FOR_TESTS: + if RUST_SLIDING_SYNC_FOR_TESTS not in ("off", "on"): + raise ValueError("SYNAPSE_TEST_RUST_SLIDING_SYNC must be off or on") + config.experimental.rust_sliding_sync = cast( + Literal["off", "on"], RUST_SLIDING_SYNC_FOR_TESTS + ) + if USE_POSTGRES_FOR_TESTS: test_db = "synapse_test_%s" % uuid.uuid4().hex @@ -1535,6 +1547,17 @@ def setup_test_homeserver( ) sync_handler._rust_sync_strict = RUST_SYNC_STRICT_FOR_TESTS + if RUST_SLIDING_SYNC_FOR_TESTS: + # Likewise for the Rust sliding sync handler. + sliding_sync_handler = hs.get_sliding_sync_handler() + if sliding_sync_handler._rust_sliding_sync is not None: + sliding_sync_handler._rust_sliding_sync = cast( + Any, _BlockingRustCalls(sliding_sync_handler._rust_sliding_sync) + ) + sliding_sync_handler._rust_sliding_sync_strict = ( + RUST_SLIDING_SYNC_STRICT_FOR_TESTS + ) + # Likewise for event loads, when they run in Rust: code that loads events # in the background would otherwise not have finished when a test checks. # `GetEventCancellationTestCase` exercises the deferreds. @@ -1597,7 +1620,7 @@ class _BlockingRustStateResolution: class _BlockingRustCalls: - """Wraps the Rust `SyncHandler` or `RustStore` so that each of its entry + """Wraps a Rust handler or the `RustStore` so that each of its entry points blocks, for the reason `_BlockingRustStateResolution` gives. Every entry point `name` that returns a Deferred has a `name_blocking` @@ -1605,7 +1628,9 @@ class _BlockingRustCalls: that twin's result in an already-fired Deferred. """ - def __init__(self, handler: "RustSyncHandler | RustStore") -> None: + def __init__( + self, handler: "RustSyncHandler | RustSlidingSyncHandler | RustStore" + ) -> None: self._handler = handler def __getattr__(self, name: str) -> Any: diff --git a/tests/utils.py b/tests/utils.py index a22da02476..8e745bf176 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -64,6 +64,14 @@ RUST_SYNC_FOR_TESTS = os.environ.get("SYNAPSE_TEST_RUST_SYNC") # Python one, fails the test, so that the suite is a differential test of the # two. Set to 0 to only count and log them, as production does. RUST_SYNC_STRICT_FOR_TESTS = os.environ.get("SYNAPSE_TEST_RUST_SYNC_STRICT", "1") != "0" +# As above, for `experimental_features.rust_sliding_sync` ("off" or "on"). +RUST_SLIDING_SYNC_FOR_TESTS = os.environ.get("SYNAPSE_TEST_RUST_SLIDING_SYNC") +# A Rust error fails the test rather than falling back to Python, so that a +# test fails on the room or request Rust got wrong. Set to 0 to fall back and +# count, as production does. +RUST_SLIDING_SYNC_STRICT_FOR_TESTS = ( + os.environ.get("SYNAPSE_TEST_RUST_SLIDING_SYNC_STRICT", "1") != "0" +) # At `rust_federation_sending: transactions`, check every transaction body Rust # builds against the one Python would build, and fail the test if they differ. RUST_FED_STRICT_FOR_TESTS = os.environ.get("SYNAPSE_TEST_RUST_FED_STRICT", "1") != "0"