mirror of
https://github.com/element-hq/synapse.git
synced 2026-11-04 09:07:22 +00:00
Port the room list filters to Rust
`rust/src/sliding_sync/room_lists/filter.rs` holds `filter_membership_for_sync` and `filter_rooms_using_tables` over `RoomForUser`, the Rust form of `RoomsForUserSlidingSync`. The caller passes the DM rooms, the user's tags and whether the connection sent a room before, which decides the state-reset keep rule. The `spaces` and `room_name_like` filters return an error, as Python raises `NotImplementedError` for them. The filtered rooms keep their input order. Python's are in set order. The differential test calls Python's method with a stand-in store over 2000 generated room sets and filters. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01L51DoxB8VnH8UjXJFDgpNG
This commit is contained in:
1 parent
b19fe79ab6
commit
c2117d3b16
5 files changed
+435
-2
No files matched your search
@@ -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)?;
|
||||
|
||||
|
||||
@@ -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:
|
||||
* <https://www.gnu.org/licenses/agpl-3.0.html>.
|
||||
*/
|
||||
|
||||
//! 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<String>,
|
||||
room_tags: &HashMap<String, HashSet<String>>,
|
||||
sent_before: impl Fn(&str) -> bool,
|
||||
) -> Result<Vec<&'a RoomForUser>, 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<PyRoomForUser>,
|
||||
filters: &str,
|
||||
dm_room_ids: HashSet<String>,
|
||||
room_tags: HashMap<String, HashSet<String>>,
|
||||
sent_before: HashSet<String>,
|
||||
) -> PyResult<Vec<String>> {
|
||||
let filters: Filters =
|
||||
serde_json::from_str(filters).map_err(|e| PyValueError::new_err(e.to_string()))?;
|
||||
let rooms: Vec<RoomForUser> = 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(())
|
||||
}
|
||||
@@ -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:
|
||||
* <https://www.gnu.org/licenses/agpl-3.0.html>.
|
||||
*/
|
||||
|
||||
//! 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<str>,
|
||||
/// `None` after a state reset.
|
||||
pub sender: Option<Arc<str>>,
|
||||
pub membership: Arc<str>,
|
||||
/// `None` after a state reset.
|
||||
pub event_id: Option<Arc<str>>,
|
||||
pub event_pos: PersistedEventPosition,
|
||||
pub room_version_id: Arc<str>,
|
||||
pub has_known_state: bool,
|
||||
pub room_type: Option<Arc<str>>,
|
||||
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>,
|
||||
String,
|
||||
Option<String>,
|
||||
i64,
|
||||
bool,
|
||||
Option<String>,
|
||||
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(())
|
||||
}
|
||||
@@ -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."""
|
||||
@@ -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)
|
||||
Reference in new issue
Block a user