diff --git a/rust/src/sliding_sync/mod.rs b/rust/src/sliding_sync/mod.rs index a831ae3146..b966beaf52 100644 --- a/rust/src/sliding_sync/mod.rs +++ b/rust/src/sliding_sync/mod.rs @@ -40,6 +40,7 @@ pub mod config; pub mod connection; pub mod request; pub mod required_state; +pub mod room_lists; /// Computes sliding sync responses with the Rust implementation, reading /// the database through the native Postgres driver. @@ -167,6 +168,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> config::register_test_functions(&child_module)?; connection::register_test_functions(&child_module)?; required_state::register_test_functions(&child_module)?; + room_lists::register_test_functions(&child_module)?; m.add_submodule(&child_module)?; diff --git a/rust/src/sliding_sync/room_lists/filter.rs b/rust/src/sliding_sync/room_lists/filter.rs new file mode 100644 index 0000000000..11186eac38 --- /dev/null +++ b/rust/src/sliding_sync/room_lists/filter.rs @@ -0,0 +1,167 @@ +/* + * 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: + * . + */ + +//! Which of a user's rooms a sync covers, and a list's filters. + +use std::collections::{HashMap, HashSet}; + +use pyo3::{exceptions::PyValueError, prelude::*, types::PyModule}; + +use crate::sliding_sync::request::Filters; +use crate::sliding_sync::room_lists::{room_from_py, PyRoomForUser, RoomForUser}; + +/// A filter Python does not implement either: `spaces` and +/// `room_name_like` raise `NotImplementedError` there. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NotImplementedFilter(pub &'static str); + +impl std::fmt::Display for NotImplementedFilter { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "the {} filter is not implemented", self.0) + } +} + +impl std::error::Error for NotImplementedFilter {} + +/// `filter_membership_for_sync`: whether the sync covers the room. Every +/// membership but a leave does, and a leave only when it is newly left or +/// a kick. A leave with no sender is a state reset. +pub fn filter_membership_for_sync(user_id: &str, room: &RoomForUser, newly_left: bool) -> bool { + &*room.membership != "leave" + || newly_left + || room + .sender + .as_deref() + .is_some_and(|sender| sender != user_id) +} + +/// `filter_rooms_using_tables`: the rooms of `rooms` that pass `filters`, +/// in their order, plus any the user was state reset out of that the +/// connection has sent before (`sent_before` is whether it has, from the +/// previous connection state). `room_tags` maps a room to the user's tags +/// on it, and is only read for the tag filters. +pub fn filter_rooms_using_tables<'a>( + rooms: &'a [RoomForUser], + filters: &Filters, + dm_room_ids: &HashSet, + room_tags: &HashMap>, + sent_before: impl Fn(&str) -> bool, +) -> Result, NotImplementedFilter> { + if filters.spaces.is_some() { + return Err(NotImplementedFilter("spaces")); + } + if filters.room_name_like.is_some() { + return Err(NotImplementedFilter("room_name_like")); + } + + let no_tags = HashSet::new(); + let passes = |room: &RoomForUser| -> bool { + if let Some(is_dm) = filters.is_dm { + if dm_room_ids.contains(&*room.room_id) != is_dm { + return false; + } + } + if let Some(is_encrypted) = filters.is_encrypted { + // An unknown encryption status fails either way. + if !room.has_known_state || room.is_encrypted != is_encrypted { + return false; + } + } + if let Some(is_invite) = filters.is_invite { + if (&*room.membership == "invite") != is_invite { + return false; + } + } + if filters.room_types.is_some() || filters.not_room_types.is_some() { + if !room.has_known_state { + return false; + } + let room_type = room.room_type.as_deref(); + if let Some(room_types) = &filters.room_types { + if !room_types.iter().any(|t| t.as_deref() == room_type) { + return false; + } + } + if let Some(not_room_types) = &filters.not_room_types { + if not_room_types.iter().any(|t| t.as_deref() == room_type) { + return false; + } + } + } + if filters.tags.is_some() || filters.not_tags.is_some() { + let tags = room_tags.get(&*room.room_id).unwrap_or(&no_tags); + if let Some(wanted) = &filters.tags { + if !wanted.iter().any(|tag| tags.contains(tag)) { + return false; + } + } + if let Some(unwanted) = &filters.not_tags { + if unwanted.iter().any(|tag| tags.contains(tag)) { + return false; + } + } + } + true + }; + + Ok(rooms + .iter() + .filter(|room| { + passes(room) + // State reset out of a room the client knows: it must hear + // of the reset whatever the filters. + || (room.event_id.is_none() && sent_before(&room.room_id)) + }) + .collect()) +} + +/// `filter_membership_for_sync` through the Rust implementation. +#[pyfunction] +fn filter_membership_for_sync_for_test( + user_id: &str, + room: PyRoomForUser, + newly_left: bool, +) -> bool { + filter_membership_for_sync(user_id, &room_from_py(room), newly_left) +} + +/// `filter_rooms_using_tables` through the Rust implementation, with +/// `filters` as JSON. Returns the room IDs that pass, or raises +/// `NotImplementedError`. +#[pyfunction] +fn filter_rooms_for_test( + rooms: Vec, + filters: &str, + dm_room_ids: HashSet, + room_tags: HashMap>, + sent_before: HashSet, +) -> PyResult> { + let filters: Filters = + serde_json::from_str(filters).map_err(|e| PyValueError::new_err(e.to_string()))?; + let rooms: Vec = rooms.into_iter().map(room_from_py).collect(); + let filtered = filter_rooms_using_tables(&rooms, &filters, &dm_room_ids, &room_tags, |r| { + sent_before.contains(r) + }) + .map_err(|e| pyo3::exceptions::PyNotImplementedError::new_err(e.to_string()))?; + Ok(filtered + .iter() + .map(|room| room.room_id.to_string()) + .collect()) +} + +pub fn register_test_functions(m: &Bound<'_, PyModule>) -> PyResult<()> { + m.add_function(wrap_pyfunction!(filter_membership_for_sync_for_test, m)?)?; + m.add_function(wrap_pyfunction!(filter_rooms_for_test, m)?)?; + Ok(()) +} diff --git a/rust/src/sliding_sync/room_lists/mod.rs b/rust/src/sliding_sync/room_lists/mod.rs new file mode 100644 index 0000000000..4fc5734f69 --- /dev/null +++ b/rust/src/sliding_sync/room_lists/mod.rs @@ -0,0 +1,81 @@ +/* + * 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 room lists: which rooms a request is interested in, filtered, +//! sorted and cut to the requested ranges +//! (`synapse.handlers.sliding_sync.room_lists`). + +use std::sync::Arc; + +use pyo3::{prelude::*, types::PyModule}; + +use crate::storage::sync_store::PersistedEventPosition; + +pub mod filter; + +/// A user's membership in a room at the `to_token`, as the sliding sync +/// tables record it: `RoomsForUserSlidingSync`, which also stands for +/// `RoomsForUser` and `RoomsForUserStateReset` where only their fields are +/// read. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RoomForUser { + pub room_id: Arc, + /// `None` after a state reset. + pub sender: Option>, + pub membership: Arc, + /// `None` after a state reset. + pub event_id: Option>, + pub event_pos: PersistedEventPosition, + pub room_version_id: Arc, + pub has_known_state: bool, + pub room_type: Option>, + pub is_encrypted: bool, +} + +/// A [`RoomForUser`] from the test form `(room_id, sender, membership, +/// event_id, stream, has_known_state, room_type, is_encrypted)`. +pub(crate) type PyRoomForUser = ( + String, + Option, + String, + Option, + i64, + bool, + Option, + bool, +); + +pub(crate) fn room_from_py(room: PyRoomForUser) -> RoomForUser { + let (room_id, sender, membership, event_id, stream, has_known_state, room_type, is_encrypted) = + room; + RoomForUser { + room_id: Arc::from(room_id), + sender: sender.map(Arc::from), + membership: Arc::from(membership), + event_id: event_id.map(Arc::from), + event_pos: PersistedEventPosition { + instance_name: Arc::from("master"), + stream, + }, + room_version_id: Arc::from("10"), + has_known_state, + room_type: room_type.map(Arc::from), + is_encrypted, + } +} + +pub fn register_test_functions(m: &Bound<'_, PyModule>) -> PyResult<()> { + filter::register_test_functions(m)?; + Ok(()) +} diff --git a/synapse/synapse_rust/sliding_sync.pyi b/synapse/synapse_rust/sliding_sync.pyi index 78dcc44c4e..1dee93c902 100644 --- a/synapse/synapse_rust/sliding_sync.pyi +++ b/synapse/synapse_rust/sliding_sync.pyi @@ -166,3 +166,19 @@ def per_connection_state_for_test( now_ms: int, ) -> tuple[bool, int, int]: """`(has_updates, len(state), len(mutable state))`.""" + +# A room for user in the test form `(room_id, sender, membership, event_id, +# stream, has_known_state, room_type, is_encrypted)`. +_PyRoom = tuple[str, str | None, str, str | None, int, bool, str | None, bool] + +def filter_membership_for_sync_for_test( + user_id: str, room: _PyRoom, newly_left: bool +) -> bool: ... +def filter_rooms_for_test( + rooms: list[_PyRoom], + filters: str, + dm_room_ids: set[str], + room_tags: dict[str, set[str]], + sent_before: set[str], +) -> list[str]: + """`filters` is JSON. Raises `NotImplementedError` as Python does.""" diff --git a/tests/handlers/test_sliding_sync_rust_pure.py b/tests/handlers/test_sliding_sync_rust_pure.py index 33c7fd42eb..fa418b1042 100644 --- a/tests/handlers/test_sliding_sync_rust_pure.py +++ b/tests/handlers/test_sliding_sync_rust_pure.py @@ -16,9 +16,10 @@ (`rust/src/sliding_sync/`) against the Python it ports, over generated inputs.""" +import json import random from types import SimpleNamespace -from typing import AbstractSet, Any, Mapping, cast +from typing import AbstractSet, Any, Coroutine, Mapping, TypeVar, cast from synapse.api.constants import EventTypes, Membership from synapse.events import EventBase @@ -28,8 +29,13 @@ from synapse.handlers.sliding_sync import ( _required_state_request, _state_filter_with_meta, ) +from synapse.handlers.sliding_sync.room_lists import ( + SlidingSyncRoomLists, + filter_membership_for_sync, +) +from synapse.storage.roommember import RoomsForUserSlidingSync from synapse.synapse_rust import sliding_sync as rust -from synapse.types import RoomStreamToken +from synapse.types import PersistedEventPosition, RoomStreamToken, StreamToken from synapse.types.handlers.sliding_sync import ( HaveSentRoom, HaveSentRoomFlag, @@ -45,6 +51,8 @@ from synapse.util.clock import Clock from tests import unittest +T = TypeVar("T") + # The values the generators draw state types and keys from: the wildcards, # local and remote users, and plain strings. _TYPES = [ @@ -593,3 +601,162 @@ class ConnectionStateTestCase(unittest.TestCase): len(python), ), ) + + +PyRoom = tuple[str, str | None, str, str | None, int, bool, str | None, bool] + + +def _run(coroutine: Coroutine[Any, Any, T]) -> T: + """Run a coroutine that never waits.""" + try: + coroutine.send(None) + except StopIteration as e: + return e.value + raise AssertionError("the coroutine waited") + + +def _random_room(rng: random.Random, room_id: str) -> PyRoom: + membership = rng.choice( + [Membership.JOIN, Membership.INVITE, Membership.LEAVE, Membership.BAN] + ) + state_reset = rng.random() < 0.1 + return ( + room_id, + None if state_reset else rng.choice([_USER_ID, "@bob:test"]), + membership, + None if state_reset else f"$event_{room_id}", + rng.randint(1, 100), + rng.random() < 0.8, + rng.choice([None, None, "m.space", "org.example.type"]), + rng.random() < 0.5, + ) + + +def _python_room(room: PyRoom) -> RoomsForUserSlidingSync: + ( + room_id, + sender, + membership, + event_id, + stream, + has_known_state, + room_type, + is_encrypted, + ) = room + return RoomsForUserSlidingSync( + room_id=room_id, + sender=sender, + membership=membership, + event_id=event_id, + event_pos=PersistedEventPosition("master", stream), + room_version_id="10", + has_known_state=has_known_state, + room_type=room_type, + is_encrypted=is_encrypted, + ) + + +def _random_filters(rng: random.Random) -> dict[str, Any]: + filters: dict[str, Any] = {} + types = [None, "m.space", "org.example.type", "org.other"] + tags = ["m.favourite", "m.lowpriority", "u.work"] + if rng.random() < 0.4: + filters["is_dm"] = rng.random() < 0.5 + if rng.random() < 0.4: + filters["is_encrypted"] = rng.random() < 0.5 + if rng.random() < 0.4: + filters["is_invite"] = rng.random() < 0.5 + if rng.random() < 0.3: + filters["room_types"] = rng.sample(types, rng.randint(0, 2)) + if rng.random() < 0.3: + filters["not_room_types"] = rng.sample(types, rng.randint(0, 2)) + if rng.random() < 0.3: + filters["tags"] = rng.sample(tags, rng.randint(0, 2)) + if rng.random() < 0.3: + filters["not_tags"] = rng.sample(tags, rng.randint(0, 2)) + if rng.random() < 0.03: + filters["spaces"] = ["!space:test"] + if rng.random() < 0.03: + filters["room_name_like"] = "name" + return filters + + +class _FakeTagStore: + def __init__(self, room_tags: dict[str, set[str]]) -> None: + self.room_tags = room_tags + + async def get_tags_for_user(self, user_id: str) -> dict[str, dict[str, Any]]: + return { + room_id: {tag: {} for tag in tags} + for room_id, tags in self.room_tags.items() + } + + +class RoomListFilterTestCase(unittest.TestCase): + """`rust/src/sliding_sync/room_lists/filter.rs`""" + + def test_filter_membership_for_sync(self) -> None: + rng = random.Random(9) + for _ in range(500): + room = _random_room(rng, "!r:test") + newly_left = rng.random() < 0.3 + self.assertEqual( + rust.filter_membership_for_sync_for_test(_USER_ID, room, newly_left), + filter_membership_for_sync( + user_id=_USER_ID, + room_membership_for_user=_python_room(room), + newly_left=newly_left, + ), + (room, newly_left), + ) + + def test_filter_rooms_using_tables(self) -> None: + rng = random.Random(10) + not_implemented = 0 + for _ in range(2000): + rooms = [_random_room(rng, room_id) for room_id in _ROOMS] + filters = _random_filters(rng) + dm_room_ids = {room_id for room_id in _ROOMS if rng.random() < 0.3} + room_tags = { + room_id: set(rng.sample(["m.favourite", "m.lowpriority", "u.work"], 1)) + for room_id in _ROOMS + if rng.random() < 0.4 + } + sent_before = {room_id for room_id in _ROOMS if rng.random() < 0.5} + + fake_room_lists = cast( + SlidingSyncRoomLists, + SimpleNamespace(store=_FakeTagStore(room_tags)), + ) + previous_state = PerConnectionState( + rooms=RoomStatusMap( + statuses={room_id: HaveSentRoom.live() for room_id in sent_before} + ) + ) + try: + python: set[str] | None = set( + _run( + SlidingSyncRoomLists.filter_rooms_using_tables( + fake_room_lists, + _USER_ID, + {room[0]: _python_room(room) for room in rooms}, + previous_state, + SlidingSyncConfig.SlidingSyncList.Filters(**filters), + cast(StreamToken, None), + dm_room_ids, + ) + ) + ) + except NotImplementedError: + python = None + not_implemented += 1 + try: + result: set[str] | None = set( + rust.filter_rooms_for_test( + rooms, json.dumps(filters), dm_room_ids, room_tags, sent_before + ) + ) + except NotImplementedError: + result = None + self.assertEqual(result, python, (rooms, filters)) + self.assertGreater(not_implemented, 0)