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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L51DoxB8VnH8UjXJFDgpNG

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<String, PyRoomConfig>,
         HashMap<String, Bound<'py, PyAny>>,
     )> {
-        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<Vec<(String, Bound<'py, PyAny>)>> {
         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::<PyResult<Vec<_>>>(),
-                    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::<PyResult<Vec<_>>>()
-                    }
-                }
+        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<T: Clone>(
+    previous: &RoomStatusMap<T>,
+    new: &MutableRoomStatusMap<T>,
+    mutable: bool,
+    room_ids: Option<&[String]>,
+) -> Vec<(String, HaveSentRoom<T>)> {
+    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<T>)>,
+    token_to_py: impl Fn(&T) -> PyResult<Bound<'py, PyAny>>,
+) -> PyResult<Vec<(String, Bound<'py, PyAny>)>> {
+    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<Vec<String>> {
     room_ids

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:
             (
                 _,

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