From 6d2032cc4f80e09dae44fe3be90037e7cbaebaec Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Oct 2026 17:05:05 +0000 Subject: [PATCH] Fix review findings on the sliding sync room lists port - Re-raise SlidingSyncUnknownPosition from finish_request. A racing request on the same connection can clear the position before the persist, and Python would fail the same way, so a Python retry only repeats the work. - Treat an empty events stream cache as a miss, as Python's get_entities_changed does, so Rust asks the database. - Copy the room statuses out of the connection state before building Python objects, so the lock is never held across a Python call. - Trace the rooms Python computes for Rust, as the Python path does. - Test the per-room fallback: Python computes a room Rust leaves to it, and its changes reach the Rust state. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01L51DoxB8VnH8UjXJFDgpNG --- rust/src/sliding_sync/state.rs | 151 ++++++++++-------- synapse/handlers/sliding_sync/__init__.py | 13 +- .../client/sliding_sync/test_rust_lists.py | 77 ++++++++- 3 files changed, 170 insertions(+), 71 deletions(-) diff --git a/rust/src/sliding_sync/state.rs b/rust/src/sliding_sync/state.rs index b6614f5f31..88a34cc9c4 100644 --- a/rust/src/sliding_sync/state.rs +++ b/rust/src/sliding_sync/state.rs @@ -38,8 +38,8 @@ use pyo3::{ use crate::caches::lock_ignoring_poison; use crate::sliding_sync::config::RoomSyncConfig; use crate::sliding_sync::connection::{ - HaveSentRoom, MutablePerConnectionState, PerConnectionState, RoomLazyMembershipChanges, - RoomStatusMap, + HaveSentRoom, MutablePerConnectionState, MutableRoomStatusMap, PerConnectionState, + RoomLazyMembershipChanges, RoomStatusMap, }; use crate::sliding_sync::result::required_state_map_from_py; use crate::sliding_sync::SlidingSyncStats; @@ -339,21 +339,27 @@ impl ConnectionState { HashMap, HashMap>, )> { - let data = lock_ignoring_poison(&self.data); - let previous = &data.previous; - let mut configs = HashMap::new(); - let mut statuses = HashMap::new(); - for room_id in room_ids { - if let Some(config) = previous.room_configs.get(&room_id) { - configs.insert(room_id.clone(), config_to_py(config)); + // Copy out under the lock, then build the Python objects without it. + let (last_used_ts, configs, sent) = { + let data = lock_ignoring_poison(&self.data); + let previous = &data.previous; + let mut configs = HashMap::new(); + let mut sent = Vec::new(); + for room_id in room_ids { + if let Some(config) = previous.room_configs.get(&room_id) { + configs.insert(room_id.clone(), config_to_py(config)); + } + let status = previous.rooms.have_sent_room(&room_id); + if !status.is_never() { + sent.push((room_id, status)); + } } - let status = previous.rooms.have_sent_room(&room_id); - if !status.is_never() { - let status = have_sent_room_to_py(py, &status, |token| token.into_pyobject(py))?; - statuses.insert(room_id, status); - } - } - Ok((previous.last_used_ts, configs, statuses)) + (previous.last_used_ts, configs, sent) + }; + let statuses = statuses_to_py(py, sent, |token| token.into_pyobject(py))? + .into_iter() + .collect(); + Ok((last_used_ts, configs, statuses)) } /// Record what Python's `get_room_sync_data` changed for the rooms it @@ -482,7 +488,8 @@ pub struct RoomStatusMapView { impl RoomStatusMapView { /// The statuses of `room_ids`, or every room's when `None`, as Python - /// `HaveSentRoom`s. + /// `HaveSentRoom`s. The statuses are copied out first, so the lock is + /// not held while calling into Python. fn statuses<'py>( &self, py: Python<'py>, @@ -490,58 +497,27 @@ impl RoomStatusMapView { ) -> PyResult)>> { let state = self.state.bind(py).get(); let instances = &state.instances; - let data = lock_ignoring_poison(&state.data); - // The flattened map, for every room, or one lookup per room. - macro_rules! collect { - ($previous:expr, $new:expr, $to_py:expr) => {{ - let to_py = $to_py; - match room_ids { - Some(room_ids) => room_ids - .iter() - .map(|room_id| { - let status = if self.mutable { - $new.have_sent_room(room_id) - } else { - $previous.have_sent_room(room_id) - }; - Ok((room_id.clone(), have_sent_room_to_py(py, &status, to_py)?)) - }) - .collect::>>(), - None => { - let flattened; - let map = if self.mutable { - flattened = $new.copy(); - &flattened - } else { - &$previous - }; - map.iter() - .map(|(room_id, status)| { - Ok((room_id.to_owned(), have_sent_room_to_py(py, status, to_py)?)) - }) - .collect::>>() - } - } + macro_rules! copy { + ($stream:ident) => {{ + let data = lock_ignoring_poison(&state.data); + copy_statuses( + &data.previous.$stream, + &data.new.$stream, + self.mutable, + room_ids, + ) }}; } match self.stream { - Stream::Rooms => collect!( - data.previous.rooms, - data.new.rooms, - |token: &RoomStreamToken| { token.into_pyobject(py) } - ), - Stream::Receipts => collect!( - data.previous.receipts, - data.new.receipts, - |token: &String| { multi_writer_to_py(py, token, instances) } - ), - Stream::AccountData => { - collect!( - data.previous.account_data, - data.new.account_data, - |token: &i64| { Ok(token.into_pyobject(py)?.into_any()) } - ) - } + Stream::Rooms => statuses_to_py(py, copy!(rooms), |token: &RoomStreamToken| { + token.into_pyobject(py) + }), + Stream::Receipts => statuses_to_py(py, copy!(receipts), |token: &String| { + multi_writer_to_py(py, token, instances) + }), + Stream::AccountData => statuses_to_py(py, copy!(account_data), |token: &i64| { + Ok(token.into_pyobject(py)?.into_any()) + }), } } @@ -556,6 +532,49 @@ impl RoomStatusMapView { } } +/// The statuses of `room_ids`, or every room's when `None`, from +/// `previous`, or from `new` when `mutable`. +fn copy_statuses( + previous: &RoomStatusMap, + new: &MutableRoomStatusMap, + mutable: bool, + room_ids: Option<&[String]>, +) -> Vec<(String, HaveSentRoom)> { + match room_ids { + Some(room_ids) => room_ids + .iter() + .map(|room_id| { + let status = if mutable { + new.have_sent_room(room_id) + } else { + previous.have_sent_room(room_id) + }; + (room_id.clone(), status) + }) + .collect(), + None if mutable => new + .copy() + .iter() + .map(|(room_id, status)| (room_id.to_owned(), status.clone())) + .collect(), + None => previous + .iter() + .map(|(room_id, status)| (room_id.to_owned(), status.clone())) + .collect(), + } +} + +fn statuses_to_py<'py, T>( + py: Python<'py>, + statuses: Vec<(String, HaveSentRoom)>, + token_to_py: impl Fn(&T) -> PyResult>, +) -> PyResult)>> { + statuses + .into_iter() + .map(|(room_id, status)| Ok((room_id, have_sent_room_to_py(py, &status, &token_to_py)?))) + .collect() +} + /// Room IDs from any iterable of strings. fn room_ids_from_py(room_ids: &Bound<'_, PyAny>) -> PyResult> { room_ids diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 98a86777da..944bf9bdbb 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -755,7 +755,9 @@ class SlidingSyncHandler: connection_position, finish_stats = await rust.finish_request( request, state, interested ) - except CancelledError: + except (CancelledError, SlidingSyncUnknownPosition): + # A request racing on the same connection cleared our position. + # Python would hit the same error, so there is nothing to retry. raise except Exception: # The persist is one transaction, so nothing was stored, and @@ -797,8 +799,11 @@ class SlidingSyncHandler: membership_changed_since_from = ( store._membership_stream_cache.has_entity_changed(user_id, from_stream) ) - changed = store._events_stream_cache.get_all_entities_changed(from_stream) - if changed.hit: + # Python's `get_entities_changed` does not trust an empty cache. + # Leaving this as None makes Rust ask the database instead. + events_cache = store._events_stream_cache + changed = events_cache.get_all_entities_changed(from_stream) + if changed.hit and events_cache._cache: events_changed_since_from = set(changed.entities) return await rust.compute_interested_rooms( request, @@ -851,6 +856,8 @@ class SlidingSyncHandler: for room_input in interested.python_room_inputs(room_ids) } + @trace + @tag_args async def handle_room(room_id: str) -> None: ( _, diff --git a/tests/rest/client/sliding_sync/test_rust_lists.py b/tests/rest/client/sliding_sync/test_rust_lists.py index 64fdbf464b..f00068fdc5 100644 --- a/tests/rest/client/sliding_sync/test_rust_lists.py +++ b/tests/rest/client/sliding_sync/test_rust_lists.py @@ -32,8 +32,16 @@ from twisted.internet.defer import Deferred from twisted.internet.testing import MemoryReactor import synapse.rest.admin -from synapse.api.constants import AccountDataTypes, EventTypes, RoomTypes -from synapse.handlers.sliding_sync import rust_sliding_sync_requests_counter +from synapse.api.constants import ( + AccountDataTypes, + EventTypes, + RelationTypes, + RoomTypes, +) +from synapse.handlers.sliding_sync import ( + rust_sliding_sync_requests_counter, + rust_sliding_sync_rooms_counter, +) from synapse.metrics import SERVER_NAME_LABEL from synapse.rest.client import account_data, knock, login, receipts, room, sync, tags from synapse.server import HomeServer @@ -49,6 +57,7 @@ from synapse.types import JsonDict, MultiWriterStreamToken, SlidingSyncStreamTok from synapse.types.handlers.sliding_sync import ( MutablePerConnectionState, SlidingSyncConfig, + StateValues, ) from synapse.util.clock import Clock @@ -436,6 +445,70 @@ class SlidingSyncRustListsTestCase(SlidingSyncBase): } ) + def test_python_rooms(self) -> None: + """Python computes the rooms Rust leaves to it on a copy of their + state, and its changes reach the Rust state.""" + user3_id = self.register_user("user3", "pass") + user3_tok = self.login(user3_id, "pass") + (rust_room,) = self._joined_rooms(1) + python_room = self.helper.create_room_as(self.user2_id, tok=self.user2_tok) + self.helper.join(python_room, self.user1_id, tok=self.user1_tok) + self.helper.join(python_room, user3_id, tok=user3_tok) + self._put_account_data( + AccountDataTypes.IGNORED_USER_LIST, {"ignored_users": {user3_id: {}}} + ) + # Rust leaves a room to Python when the latest reply in a thread on + # its timeline is from an ignored user. + root_id = self._send_message(python_room) + self.helper.send_event( + python_room, + EventTypes.Message, + { + "msgtype": "m.text", + "body": "reply", + "m.relates_to": {"rel_type": RelationTypes.THREAD, "event_id": root_id}, + }, + tok=user3_tok, + ) + self._send_message(rust_room) + sync_body: JsonDict = { + "lists": { + "all": { + "ranges": [[0, 99]], + "required_state": [ + [EventTypes.Create, ""], + [EventTypes.Member, StateValues.LAZY], + ], + "timeline_limit": 3, + } + } + } + fallback = rust_sliding_sync_rooms_counter.labels( + result="fallback", **{SERVER_NAME_LABEL: self.hs.hostname} + ) + before = fallback._value.get() + response = self._sync_both(sync_body) + self.assertEqual(fallback._value.get() - before, 1) + self.assertEqual(set(response["rooms"]), {rust_room, python_room}) + + # The next requests start from the state Python's room wrote. + self._send_message(python_room) + self._send_message(rust_room) + self._sync_both(sync_body) + before = fallback._value.get() + self._sync_both( + { + "lists": { + "all": { + **sync_body["lists"]["all"], + "required_state": [[EventTypes.Member, "*"]], + "timeline_limit": 5, + } + } + } + ) + self.assertEqual(fallback._value.get() - before, 1) + def test_extensions_state(self) -> None: """The receipts and account data extensions read and update the Rust connection state through its views: rooms sent before, rooms with