diff --git a/rust/src/events/filter.rs b/rust/src/events/filter.rs index 7e39972c62..7f26f3042d 100644 --- a/rust/src/events/filter.rs +++ b/rust/src/events/filter.rs @@ -33,12 +33,13 @@ pub fn event_visible_to_server_py( memberships: Vec<(String, String)>, // (state_key, membership) ) -> PyResult { event_visible_to_server( - sender, - target_server_name, - history_visibility, - erased_senders, + &target_server_name, + &history_visibility, + erased_senders.get(&sender).copied().unwrap_or(false), partial_state_invisible, - memberships, + memberships + .iter() + .map(|(state_key, membership)| (state_key.as_str(), membership.as_str())), ) .map_err(|e| PyValueError::new_err(format!("{e}"))) } @@ -53,18 +54,19 @@ pub fn event_visible_to_server_py( /// - E was created by this homeserver, AND: /// - the partial state at E has world readable or shared history vis, OR /// - the partial state at E says that the target server is in the room. -pub fn event_visible_to_server( - sender: String, - target_server_name: String, - history_visibility: String, - erased_senders: HashMap, +/// +/// `sender_erased` is whether the sender has asked to be erased, and +/// `memberships` the `(state_key, membership)` at E of the target server's +/// users. +pub fn event_visible_to_server<'a>( + target_server_name: &str, + history_visibility: &str, + sender_erased: bool, partial_state_invisible: bool, - memberships: Vec<(String, String)>, // (state_key, membership) + memberships: impl IntoIterator, ) -> anyhow::Result { - if let Some(&erased) = erased_senders.get(&sender) { - if erased { - return Ok(false); - } + if sender_erased { + return Ok(false); } if partial_state_invisible { @@ -79,7 +81,7 @@ pub fn event_visible_to_server( let mut visible = false; for (state_key, membership) in memberships { - let state_key = UserID::try_from(state_key.as_ref()) + let state_key = UserID::try_from(state_key) .map_err(|e| anyhow::anyhow!(format!("invalid user_id ({state_key}): {e}")))?; if state_key.server_name() != target_server_name { return Err(anyhow::anyhow!( @@ -88,7 +90,7 @@ pub fn event_visible_to_server( )); } - match membership.as_str() { + match membership { MEMBERSHIP_INVITE => { if history_visibility == HISTORY_VISIBILITY_INVITED { visible = true; diff --git a/rust/src/storage/sync_store.rs b/rust/src/storage/sync_store.rs index 798a9f2fd0..38ba249183 100644 --- a/rust/src/storage/sync_store.rs +++ b/rust/src/storage/sync_store.rs @@ -679,6 +679,33 @@ impl<'a> SyncReader<'a> { Ok(result) } + /// `get_partial_state_events`, less the cache: which of the events + /// have only partial state, from a faster join. + pub async fn partial_state_events( + &self, + event_ids: &[Arc], + ) -> anyhow::Result>> { + let mut partial = HashSet::new(); + if event_ids.is_empty() { + return Ok(partial); + } + let borrowed = self.borrow().await?; + for batch in event_ids.chunks(SELECT_MANY_BATCH_SIZE) { + let batch: Vec<&str> = batch.iter().map(|id| &**id).collect(); + let rows = borrowed + .query( + "SELECT event_id FROM partial_state_events WHERE event_id = ANY($1)", + &[&batch], + ) + .await?; + for row in rows { + let event_id: String = row.try_get(0)?; + partial.insert(Arc::from(event_id)); + } + } + Ok(partial) + } + /// `is_partial_state_room`. pub async fn is_partial_state_room(&self, room_id: &str) -> anyhow::Result { let rooms = self.partial_state_rooms(&[Arc::from(room_id)]).await?; @@ -2170,6 +2197,40 @@ impl PySyncStoreForTest { } } + /// [`crate::visibility::filter_events_for_server`]. `None` where Python + /// would wait for the full state. + #[pyo3(signature = (target_server_name, local_server_name, events, *, redact, filter_out_erased_senders, filter_out_remote_partial_state_events))] + #[allow(clippy::too_many_arguments)] + fn filter_events_for_server( + &self, + py: Python<'_>, + target_server_name: String, + local_server_name: String, + events: Vec>, + redact: bool, + filter_out_erased_senders: bool, + filter_out_remote_partial_state_events: bool, + ) -> PyResult>> { + let events: Vec = events.iter().map(|e| e.get().shallow_clone()).collect(); + let filtered = self.run(py, |reader| { + async move { + let filter = crate::visibility::ServerFilter { + target_server_name: &target_server_name, + local_server_name: &local_server_name, + redact, + filter_out_erased_senders, + filter_out_remote_partial_state_events, + }; + crate::visibility::filter_events_for_server(&reader, &filter, events).await + } + .boxed() + })?; + Ok(match filtered { + crate::visibility::ServerFiltered::Events(events) => Some(events), + crate::visibility::ServerFiltered::NeedsFullState => None, + }) + } + fn rooms_for_user(&self, py: Python<'_>, user_id: String) -> PyResult> { let rooms = self.run(py, |reader| { async move { reader.rooms_for_user(&user_id).await }.boxed() diff --git a/rust/src/visibility.rs b/rust/src/visibility.rs index bb8a9eed7d..7c58253d22 100644 --- a/rust/src/visibility.rs +++ b/rust/src/visibility.rs @@ -21,8 +21,16 @@ //! job, so that this can serve the sync handler and any later port alike. //! [`check_event_for_client`] is the whole per-event decision; the //! functions below it are the steps, named as the Python ones are. +//! +//! [`filter_events_for_server`] is the server half, +//! `synapse.visibility.filter_events_for_server`. It loads what it needs +//! itself, through a [`SyncReader`], and leaves the per-event decision to +//! [`event_visible_to_server`]. -use std::{collections::HashSet, sync::Arc}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; use pyo3::{ prelude::*, @@ -36,8 +44,11 @@ use crate::events::{ join_rules_field::JOIN_RULE, membership_field::MEMBERSHIP, }, + filter::event_visible_to_server, redact_event_py, Event, }; +use crate::state::StateKey; +use crate::storage::{events::RedactBehaviour, sync_store::SyncReader}; /// `EventTypes.Dummy`. const M_DUMMY: &str = "org.matrix.dummy_event"; @@ -400,6 +411,207 @@ pub fn keep_soft_failed(event: &Event, policy: Option<&SoftFailedPolicy>) -> boo } } +// The server half. + +/// What [`filter_events_for_server`] is asked to do: its arguments, less +/// the events. +#[derive(Debug, Clone, Copy)] +pub struct ServerFilter<'a> { + pub target_server_name: &'a str, + pub local_server_name: &'a str, + /// Include the redacted form of an event the server may not see, rather + /// than leave it out. + pub redact: bool, + /// Also leave out the events of senders who asked to be erased. + pub filter_out_erased_senders: bool, + /// Also leave out remote events in rooms we only have partial state + /// for. + pub filter_out_remote_partial_state_events: bool, +} + +/// What [`filter_events_for_server`] found. +pub enum ServerFiltered { + Events(Vec), + /// An event needs the memberships at it and has only partial state. + /// Python waits for the full state there. We do not, and leave it to + /// the caller. + NeedsFullState, +} + +/// The server name of a user ID, as `get_domain_from_id`. +fn domain_of(user_id: &str) -> anyhow::Result<&str> { + user_id + .split_once(':') + .map(|(_, domain)| domain) + .ok_or_else(|| anyhow::anyhow!("Invalid ID: {user_id}")) +} + +/// `filter_events_for_server`: the events the target server may see, in +/// order. Each event it may not see is left out, or redacted if +/// `filter.redact`. +pub async fn filter_events_for_server( + reader: &SyncReader<'_>, + filter: &ServerFilter<'_>, + events: Vec, +) -> anyhow::Result { + let erased = if filter.filter_out_erased_senders { + let mut senders: Vec> = events.iter().map(|e| Arc::from(e.sender())).collect(); + senders.sort(); + senders.dedup(); + reader.are_users_erased(&senders).await? + } else { + HashMap::new() + }; + + // Python checks this first, so that a room fully joined while we look + // cannot let its events through on an old view of the memberships. + let mut partial_state_invisible: HashSet<&str> = HashSet::new(); + if filter.filter_out_remote_partial_state_events { + for event in &events { + if domain_of(event.sender())? != filter.local_server_name + && reader.is_partial_state_room(event.room_id()).await? + { + partial_state_invisible.insert(event.event_id()); + } + } + } + + // `_event_to_history_vis`. An outlier has no state, and counts as + // shared. + let mut with_state: Vec> = Vec::with_capacity(events.len()); + for event in &events { + if !event.internal_metadata().is_outlier()? { + with_state.push(Arc::from(event.event_id())); + } + } + let event_to_group = reader.state_group_for_events(&with_state).await?; + if let Some(missing) = with_state.iter().find(|id| !event_to_group.contains_key(*id)) { + anyhow::bail!("No state group for event {missing}"); + } + let mut groups: Vec = event_to_group.values().copied().collect(); + groups.sort_unstable(); + groups.dedup(); + let group_state = reader.state_for_groups(&groups).await?; + let state_of = |event_id: &str| { + event_to_group + .get(event_id) + .and_then(|group| group_state.get(group)) + }; + + let vis_key = StateKey::new(M_ROOM_HISTORY_VISIBILITY, ""); + let mut vis_ids: Vec> = group_state + .values() + .filter_map(|state| state.get(&vis_key).cloned()) + .collect(); + vis_ids.sort(); + vis_ids.dedup(); + let vis_events = reader + .events(&vis_ids, RedactBehaviour::Redact, false, false) + .await?; + let vis_events: HashMap<&str, &Event> = vis_events + .iter() + .map(|event| (event.event_id(), event)) + .collect(); + + let mut history_visibility: HashMap<&str, &str> = HashMap::with_capacity(events.len()); + for event in &events { + let mut vis = SHARED; + if let Some(vis_id) = state_of(event.event_id()).and_then(|state| state.get(&vis_key)) { + let vis_event = vis_events + .get(&**vis_id) + .ok_or_else(|| anyhow::anyhow!("Missing history visibility event {vis_id}"))?; + if let Some(value) = vis_event.content().get_field(HISTORY_VISIBILITY) { + vis = value.as_str().ok_or_else(|| { + anyhow::anyhow!("Invalid history visibility in {vis_id}") + })?; + } + } + history_visibility.insert(event.event_id(), vis); + } + + // `_event_to_memberships`, for the events with restricted visibility: + // the memberships at each of the target server's users. + let restricted: Vec> = events + .iter() + .filter(|event| { + !matches!( + history_visibility[event.event_id()], + SHARED | WORLD_READABLE + ) + }) + .map(|event| Arc::from(event.event_id())) + .collect(); + let mut member_events: HashMap, Event> = HashMap::new(); + if !restricted.is_empty() { + // Python waits for the full state of these. + if !reader.partial_state_events(&restricted).await?.is_empty() { + return Ok(ServerFiltered::NeedsFullState); + } + let mut member_ids: Vec> = restricted + .iter() + .filter_map(|id| state_of(id)) + .flat_map(|state| state.iter()) + .filter(|(key, _)| is_target_member(key, filter.target_server_name)) + .map(|(_, event_id)| Arc::clone(event_id)) + .collect(); + member_ids.sort(); + member_ids.dedup(); + for event in reader + .events(&member_ids, RedactBehaviour::Redact, false, false) + .await? + { + member_events.insert(Arc::from(event.event_id()), event); + } + } + + let mut visible = Vec::with_capacity(events.len()); + for event in &events { + let event_id = event.event_id(); + let mut memberships: Vec<(&str, &str)> = Vec::new(); + if restricted.iter().any(|id| &**id == event_id) { + for (key, member_id) in state_of(event_id).into_iter().flat_map(|state| state.iter()) { + if !is_target_member(key, filter.target_server_name) { + continue; + } + let Some(member) = member_events.get(member_id) else { + continue; + }; + let membership = membership_of(member).ok_or_else(|| { + anyhow::anyhow!("Membership event {member_id} has no membership") + })?; + memberships.push((&key.1, membership)); + } + } + let sender_erased = erased.get(event.sender()).copied().unwrap_or(false); + visible.push(event_visible_to_server( + filter.target_server_name, + history_visibility[event_id], + sender_erased, + partial_state_invisible.contains(event_id), + memberships, + )?); + } + + let mut filtered = Vec::with_capacity(events.len()); + for (event, visible) in events.into_iter().zip(visible) { + if visible { + filtered.push(event); + } else if filter.redact { + filtered.push(redact_event_py(&event).map_err(|err| anyhow::anyhow!("{err}"))?); + } + } + Ok(ServerFiltered::Events(filtered)) +} + +/// Whether `key` is the membership of one of `server_name`'s users. +fn is_target_member(key: &StateKey, server_name: &str) -> bool { + &*key.0 == M_ROOM_MEMBER + && key + .1 + .split_once(':') + .is_some_and(|(_, domain)| domain == server_name) +} + // The entry point for the differential test. /// [`check_event_for_client`] through the Rust implementation. `state` maps diff --git a/synapse/synapse_rust/sync.pyi b/synapse/synapse_rust/sync.pyi index 05952ff8d4..01c03fa402 100644 --- a/synapse/synapse_rust/sync.pyi +++ b/synapse/synapse_rust/sync.pyi @@ -374,6 +374,19 @@ class SyncStoreForTest: Rust sync handler's caches. For `tests/storage/test_sync_store.py`.""" def __init__(self, pool: ConnectionPool, handler: SyncHandler) -> None: ... + def filter_events_for_server( + self, + target_server_name: str, + local_server_name: str, + events: Sequence[Event], + *, + redact: bool, + filter_out_erased_senders: bool, + filter_out_remote_partial_state_events: bool, + ) -> list[Event] | None: + """`synapse.visibility.filter_events_for_server` in Rust. `None` where + Python would wait for the full state.""" + def rooms_for_user(self, user_id: str) -> set[str]: """`get_rooms_for_user`.""" diff --git a/tests/test_rust_server_visibility.py b/tests/test_rust_server_visibility.py new file mode 100644 index 0000000000..e35d1aa016 --- /dev/null +++ b/tests/test_rust_server_visibility.py @@ -0,0 +1,220 @@ +# +# 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 Rust `filter_events_for_server` (`rust/src/visibility.rs`) against the +Python one, on the same rooms. These need the native Rust Postgres driver: + + SYNAPSE_TEST_RUST_DRIVER=1 POSTGRES=1 run-trial.sh tests.test_rust_server_visibility +""" + +import itertools +import random +from typing import Any +from unittest import SkipTest + +from parameterized import parameterized + +from twisted.internet.testing import MemoryReactor + +from synapse.api.room_versions import RoomVersions +from synapse.events import EventBase +from synapse.events.snapshot import EventContext +from synapse.server import HomeServer +from synapse.storage.rust_pool import RustConnectionPool +from synapse.synapse_rust.sync import SyncStoreForTest +from synapse.util.clock import Clock +from synapse.visibility import filter_events_for_server + +from tests import unittest +from tests.test_utils.event_injection import inject_member_event +from tests.test_visibility import inject_message_event, inject_visibility_event +from tests.utils import create_room + +ROOM_ID = "!test:local" +LOCAL = "local" +SERVERS = ["a", "b", "c"] +VISIBILITIES = ["world_readable", "shared", "invited", "joined"] +MEMBERSHIPS = ["join", "invite", "leave", "ban", "knock"] + +# Every combination of the three flags. +FLAGS = [ + { + "redact": redact, + "filter_out_erased_senders": erased, + "filter_out_remote_partial_state_events": partial, + } + for redact, erased, partial in itertools.product((False, True), repeat=3) +] + + +class RustServerVisibilityTestCase(unittest.HomeserverTestCase): + def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: + hs = self.setup_test_homeserver() + if not isinstance( + hs.get_datastores().main.db_pool._db_pool, RustConnectionPool + ): + raise SkipTest("needs the native Rust Postgres driver") + return hs + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = hs.get_datastores().main + self.storage = hs.get_storage_controllers() + db_pool = self.store.db_pool._db_pool + assert isinstance(db_pool, RustConnectionPool) + assert db_pool._pool is not None + rust_handler = hs.get_rust_handlers().sync + assert rust_handler is not None + self.rust = SyncStoreForTest(db_pool._pool, rust_handler) + + self.get_success(create_room(hs, ROOM_ID, f"@creator:{LOCAL}")) + + def _compare( + self, target: str, events: list[EventBase], flags: dict[str, Any] + ) -> list[EventBase]: + """Filter with both, check they agree, and return the result.""" + python = self.get_success( + filter_events_for_server(self.storage, target, LOCAL, events, **flags) + ) + rust = self.rust.filter_events_for_server(target, LOCAL, events, **flags) + assert rust is not None, "Rust wanted to wait for the full state" + self.assertEqual( + [event.get_pdu_json() for event in rust], + [event.get_pdu_json() for event in python], + f"target={target} flags={flags}", + ) + self.assertEqual( + [event.internal_metadata.is_redacted() for event in rust], + [event.internal_metadata.is_redacted() for event in python], + ) + return python + + def _random_room(self, seed: int) -> list[EventBase]: + """A room history of visibility changes, memberships of users on + several servers and messages, in a random order.""" + rng = random.Random(seed) + users = [f"@u{i}:{server}" for server in SERVERS + [LOCAL] for i in range(2)] + events: list[EventBase] = [] + for _ in range(40): + action = rng.random() + if action < 0.15: + event = self.get_success( + inject_visibility_event( + self.hs, ROOM_ID, rng.choice(users), rng.choice(VISIBILITIES) + ) + ) + elif action < 0.5: + event = self.get_success( + inject_member_event( + self.hs, ROOM_ID, rng.choice(users), rng.choice(MEMBERSHIPS) + ) + ) + else: + event = self.get_success( + inject_message_event(self.hs, ROOM_ID, rng.choice(users)) + ) + events.append(event) + + for user in rng.sample(users, 2): + self.get_success(self.store.mark_user_erased(user)) + return events + + @parameterized.expand([(seed,) for seed in range(6)]) + def test_random_rooms(self, seed: int) -> None: + events = self._random_room(seed) + rng = random.Random(seed) + outcomes = set() + for target in SERVERS + ["nobody"]: + batch = rng.sample(events, 15) + for flags in FLAGS: + filtered = self._compare(target, batch, flags) + outcomes.add( + "omitted" + if len(filtered) < len(batch) + else "redacted" + if any(e.internal_metadata.is_redacted() for e in filtered) + else "all" + ) + # The rooms are random, but each should exercise the filter. + self.assertIn("omitted", outcomes) + self.assertIn("redacted", outcomes) + + def test_outlier(self) -> None: + """An outlier has no state, so any server may see it.""" + self.get_success( + inject_visibility_event(self.hs, ROOM_ID, f"@creator:{LOCAL}", "joined") + ) + message = self.get_success( + inject_message_event(self.hs, ROOM_ID, f"@creator:{LOCAL}") + ) + outlier = self._inject_outlier() + for flags in FLAGS: + filtered = self._compare("a", [outlier, message], flags) + self.assertEqual(filtered[0].event_id, outlier.event_id) + + def test_partial_state_room(self) -> None: + """Remote events in a room with partial state are left out, and an + event that needs the memberships waits for the full state.""" + self.get_success( + inject_visibility_event(self.hs, ROOM_ID, f"@creator:{LOCAL}", "joined") + ) + self.get_success(inject_member_event(self.hs, ROOM_ID, "@u:a", "join")) + remote = self.get_success(inject_message_event(self.hs, ROOM_ID, "@u:a")) + local = self.get_success( + inject_message_event(self.hs, ROOM_ID, f"@creator:{LOCAL}") + ) + + self.get_success( + self.store.db_pool.simple_insert( + "partial_state_rooms", {"room_id": ROOM_ID} + ) + ) + self.store.is_partial_state_room.invalidate((ROOM_ID,)) + for flags in FLAGS: + self._compare("a", [remote, local], flags) + + self.get_success( + self.store.db_pool.simple_insert( + "partial_state_events", {"room_id": ROOM_ID, "event_id": local.event_id} + ) + ) + self.assertIsNone( + self.rust.filter_events_for_server( + "a", + LOCAL, + [local], + redact=False, + filter_out_erased_senders=True, + filter_out_remote_partial_state_events=False, + ) + ) + + def _inject_outlier(self) -> EventBase: + builder = self.hs.get_event_builder_factory().for_room_version( + RoomVersions.V1, + { + "type": "m.room.member", + "sender": "@test:user", + "state_key": "@test:user", + "room_id": ROOM_ID, + "content": {"membership": "join"}, + }, + ) + event = self.get_success(builder.build(prev_event_ids=[], auth_event_ids=[])) + event.internal_metadata.outlier = True + persistence = self.storage.persistence + assert persistence is not None + self.get_success( + persistence.persist_event(event, EventContext.for_outlier(self.storage)) + ) + return event