Port filter_events_for_server's loading to Rust

`visibility::filter_events_for_server` reads the erased senders, the
partial state rooms, the history visibility and the target server's
memberships at each event through a `SyncReader`, and leaves the
per-event decision to `event_visible_to_server`. That now borrows its
arguments.

Python waits for the full state of an event whose memberships it needs.
Rust cannot, and returns `NeedsFullState` for the caller to handle.

The differential tests compare it with Python's on random rooms, an
outlier and a room with partial state, for every combination of flags.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01R5b7Ruy4cgt27Mjq6Ptxbb

rust/src/events/filter.rs

@@ -33,12 +33,13 @@ pub fn event_visible_to_server_py(
     memberships: Vec<(String, String)>, // (state_key, membership)
 ) -> PyResult<bool> {
     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<String, bool>,
+///
+/// `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<Item = (&'a str, &'a str)>,
 ) -> anyhow::Result<bool> {
-    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;

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<str>],
+    ) -> anyhow::Result<HashSet<Arc<str>>> {
+        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<bool> {
         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<Bound<'_, Event>>,
+        redact: bool,
+        filter_out_erased_senders: bool,
+        filter_out_remote_partial_state_events: bool,
+    ) -> PyResult<Option<Vec<Event>>> {
+        let events: Vec<Event> = 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<HashSet<String>> {
         let rooms = self.run(py, |reader| {
             async move { reader.rooms_for_user(&user_id).await }.boxed()

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<Event>),
+    /// 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<Event>,
+) -> anyhow::Result<ServerFiltered> {
+    let erased = if filter.filter_out_erased_senders {
+        let mut senders: Vec<Arc<str>> = 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<Arc<str>> = 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<i64> = 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<Arc<str>> = 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<Arc<str>> = 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<Arc<str>, 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<Arc<str>> = 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

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`."""
 

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+"""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