C3: Serve MSC4430 member key rooms to clients

What a client sees of a member key room: every event names a member key as
its sender and, through MSC4428, as its stable ID; `member_info` resolves
the keys to users; the requester is recognised by their key wherever the
server decides what they may see or which events are theirs; and `/members`
and `/joined_members` are keyed by key. One homeserver only, but the code
answers for members of other servers too (a verified link names the user it
claims, anything else is nobody).

`member_info`. MSC4428's map, which only named erased users (the stable ID
of an ordinary room is the user ID and the entry is omitted), now gives
every member key an entry: the user behind a key we hold, the user named by
a verified link for another server's key, and `null` for an erased user or a
key nobody vouches for (an unverified claim is anyone's to make). The
resolution is one batched store lookup (`get_user_ids_for_identities`,
exposed by the member key handler as `user_ids_for_identities`), so the map
costs rooms without member keys nothing. The map already covers `/sync`
(full and incremental), sliding sync, `/members` and `/joined_members`,
which are keyed by state key and so by key; nothing there changes. An
erasure is recorded on the change stream under the stable ID that changed,
which in these rooms is the member's key, so the entry that flips to `null`
is the one clients hold.

Visibility. `filter_and_transform_events_for_client` decided what the
requester may see from their membership under their user ID, which the state
of a member key room does not hold: a joined user saw nothing. It now looks
the requester up under their key in each member key room the batch touches,
for both the allow check and the membership the event is annotated with, and
maps senders back to users, per room (a key is only a user in the room it
was looked up in), for the two per-user checks: erased senders, whose events
are pruned for users who joined later, so a key with no attributable user is
not erased; and ignored senders, where a member may be ignored as the user
behind their key or as the key itself, since a key with no attributable user
is the only handle a client has on it.
`filter_event_for_clients_with_state`, which push evaluation uses to decide
who may see an event, does the same per local user, for the ignore lists as
for the memberships, including for the out-of-band membership events it
returns to their target. The push evaluator itself resolves the sender (so a
user is not notified of their own events), the invitee and the inviter (so
an invite notifies its target, and the invitee's invite rules see the
inviter as a user, where one is attributable).

The requester's own events. The serializer echoes `transaction_id` and
`delay_id` only to the sender's own session, by comparing the sender with
the requester's user ID, which in a member key room never matches.
`SerializeEventConfig.requester_member_keys` (a new keyword-only argument,
`None` by default, so existing constructions are unchanged) carries the
requester's keys and the Rust serializer accepts a sender among them;
`create_config` fills it from the cached per-user keys when the feature is
on. Keys are unique across rooms, so a key of the requester's can only ever
be their own sender.

Sync. Where the handlers compared the requester's user ID with a sender or
a membership state key, they now use the identity the room knows them by
(`room_identity_for_user`: their key where they have one, their user ID
otherwise, or the cached map of their keys per room): the incremental sync
reads the requester's previous membership under their key, so a change to
their own membership (a display name) is an ordinary timeline event rather
than the room sent again as newly joined, and a leave is newly left; the
heroes of a nameless room no longer include the requester; the lazy-loading
full sync fetches the requester's own membership under their key; a leave
the user sent themselves is told from a kick by their keys per room, in
sync and in the sliding sync room lists; the join-race guard asks the
user-keyed membership table; and sliding sync's `$ME` is the requester's
key, including in the computation of which required state has changed
between requests. An invite's sender is resolved to a user (where
attributable) for the ignore list and the invite rules, in sync and in
both sliding sync room lists: a key on the ignore list hides the invite
too, and an unattributable key no longer reaches the invite rules, which
parse user IDs and would have failed the whole sync. The members handed to
presence and to device lists, by sync, by the device handler for newly
joined and left rooms, and by the room initial sync, are mapped to users
first, so clients are told about users' devices and presence, not keys'.
The three storage queries that find a user's membership changes in
`current_state_delta_stream` and `room_memberships` share one clause that,
with the feature on, also matches the keys we hold for the user, each
scoped to its room (keys are public strings anyone can name as a state key
elsewhere) and spelled out from the cached per-user keys so that every arm
of the clause can use the tables' indexes; a user with no keys, and the
feature being off, get the legacy clause byte for byte. Sliding sync's
"newly joined" and "newly left" detection, and the device handler's
shared-room tracking, run on those queries.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

rust/src/events/serialize.rs

@@ -28,7 +28,7 @@
 //! batch and then serializes each event — recursing entirely in Rust into
 //! redactions and bundled aggregations via [`serialize_event`].
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use pyo3::{
     exceptions::{PyTypeError, PyValueError},
@@ -132,12 +132,29 @@ pub struct SerializeEventConfig {
     /// `true`, the sender is repeated in `unsigned` as
     /// `org.matrix.msc4428.stable_id`.
     msc4428_enabled: bool,
+    /// MSC4430: the requester's member keys. In a member key room the `sender`
+    /// of the requester's own events is their key in that room rather than
+    /// their user ID, so sender-only fields (`transaction_id`, `delay_id`)
+    /// are also gated on these.
+    requester_member_keys: Option<HashSet<String>>,
 }
 
 #[pymethods]
 impl SerializeEventConfig {
     #[new]
     #[allow(clippy::too_many_arguments)]
+    #[pyo3(signature = (
+        *,
+        as_client_event,
+        event_format,
+        requester,
+        event_field_allowlist,
+        include_stripped_room_state,
+        include_admin_metadata,
+        msc4354_enabled,
+        msc4428_enabled,
+        requester_member_keys = None,
+    ))]
     fn new(
         as_client_event: bool,
         event_format: EventFormat,
@@ -147,8 +164,10 @@ impl SerializeEventConfig {
         include_admin_metadata: bool,
         msc4354_enabled: bool,
         msc4428_enabled: bool,
+        requester_member_keys: Option<Vec<String>>,
     ) -> PyResult<Self> {
         let requester = requester.map(|r| r.get().clone());
+        let requester_member_keys = requester_member_keys.map(HashSet::from_iter);
 
         Ok(Self {
             as_client_event,
@@ -159,6 +178,7 @@ impl SerializeEventConfig {
             include_admin_metadata,
             msc4354_enabled,
             msc4428_enabled,
+            requester_member_keys,
         })
     }
 }
@@ -437,7 +457,12 @@ fn serialize_event_value(
     // was sent by the same session (or, where appropriate, the same sender) as
     // the one requesting the event.
     if let Some(requester) = &config.requester {
-        if requester.user_id == event.sender() {
+        let sent_by_requester = requester.user_id == event.sender()
+            || config
+                .requester_member_keys
+                .as_ref()
+                .is_some_and(|keys| keys.contains(event.sender()));
+        if sent_by_requester {
             if let Some(txn_id) = event.internal_metadata.txn_id()? {
                 if let Some(event_device_id) = event.internal_metadata.device_id()? {
                     if Some(event_device_id.as_str()) == requester.device_id.as_deref() {

synapse/events/utils.py

@@ -183,7 +183,8 @@ class EventClientSerializer:
         Helper method that sets the `include_admin_metadata` field based on
         whether the requester is a server admin if it is not explicitly
         provided. Also sets the `msc4354_enabled` and `msc4428_enabled` fields
-        based on the homeserver config.
+        based on the homeserver config, and, with MSC4430 on, the requester's
+        member keys (the senders of their own events in member key rooms).
 
         Args:
             as_client_event: Whether to serialize the events as client events.
@@ -210,6 +211,14 @@ class EventClientSerializer:
             else:
                 include_admin_metadata = False
 
+        requester_member_keys: list[str] | None = None
+        if requester is not None and self._config.experimental.msc4430_enabled:
+            keys_by_room = await self._store.get_local_member_keys_for_user(
+                requester.user.to_string()
+            )
+            if keys_by_room:
+                requester_member_keys = list(keys_by_room.values())
+
         return SerializeEventConfig(
             as_client_event=as_client_event,
             event_format=event_format,
@@ -219,6 +228,7 @@ class EventClientSerializer:
             include_admin_metadata=include_admin_metadata,
             msc4354_enabled=self._config.experimental.msc4354_enabled,
             msc4428_enabled=self._config.experimental.msc4428_enabled,
+            requester_member_keys=requester_member_keys,
         )
 
     async def serialize_event(

synapse/handlers/device.py

@@ -717,6 +717,15 @@ class DeviceHandler:
 
         return device_list_updates
 
+    async def _users_behind_members(
+        self, room_id: str, members: Iterable[str]
+    ) -> set[str]:
+        """The users behind a room's membership state keys: the keys themselves
+        in rooms without member keys; in an MSC4430 member key room, the users
+        the keys map to (keys that map to nobody are left out)."""
+        mapped = await self.store.get_user_ids_for_identities(room_id, members)
+        return {user_id for user_id in mapped.values() if user_id is not None}
+
     async def generate_sync_entry_for_device_list(
         self,
         user_id: str,
@@ -774,7 +783,10 @@ class DeviceHandler:
         # Step 1b, check for newly joined rooms
         for room_id in newly_joined_rooms:
             joined_users = await self.store.get_users_in_room(room_id)
-            newly_joined_or_invited_or_knocked_users.update(joined_users)
+            # (Member keys in an MSC4430 room; device lists are about users.)
+            newly_joined_or_invited_or_knocked_users.update(
+                await self._users_behind_members(room_id, joined_users)
+            )
 
         # TODO: Check that these users are actually new, i.e. either they
         # weren't in the previous sync *or* they left and rejoined.
@@ -788,7 +800,9 @@ class DeviceHandler:
         # Now find users that we no longer track
         for room_id in newly_left_rooms:
             left_users = await self.store.get_users_in_room(room_id)
-            newly_left_users.update(left_users)
+            newly_left_users.update(
+                await self._users_behind_members(room_id, left_users)
+            )
 
         # Remove any users that we still share a room with.
         left_users_rooms = await self.store.get_rooms_for_users(newly_left_users)

synapse/handlers/initial_sync.py

@@ -466,7 +466,14 @@ class InitialSyncHandler:
             if not self.hs.config.server.presence_enabled:
                 return []
 
-            states = await presence_handler.get_states([m.sender for m in room_members])
+            # (The members as users: in an MSC4430 member key room the
+            # senders are their keys.)
+            member_users = await self.store.get_user_ids_for_identities(
+                room_id, [m.sender for m in room_members]
+            )
+            states = await presence_handler.get_states(
+                [u for u in member_users.values() if u is not None]
+            )
 
             return [
                 {

synapse/handlers/member_info.py

@@ -13,11 +13,14 @@
 #
 """MSC4428: the per-room `member_info` map.
 
-`member_info` maps a room member's stable ID (today: their user ID, carried on
-events as `unsigned.org.matrix.msc4428.stable_id`) to the user ID it currently
-stands for, which may be `None` once the user is erased. Entries whose user ID
-equals the stable ID are omitted, as the MSC allows, so in today's rooms the map
-only ever names erased users.
+`member_info` maps a room member's stable ID (carried on events as
+`unsigned.org.matrix.msc4428.stable_id`) to the user ID it currently stands
+for, which may be `None`. In rooms without member keys the stable ID is the
+user ID itself and the entry is omitted, as the MSC allows, unless the user is
+erased. In MSC4430 member key rooms the stable ID is the member key, and every
+key gets an entry: the user behind a key we hold (`None` once erased), the user
+a verified link names for another server's key, and `None` for a key whose
+link is unverified or unknown.
 
 Most changes to the map come with an event in the room (a join, a leave) and
 are delivered alongside it. Changes without such an anchor, erasure today, are
@@ -41,6 +44,7 @@ class MemberInfoHandler:
         self._store = hs.get_datastores().main
         self._state_storage_controller = hs.get_storage_controllers().state
         self._notifier = hs.get_notifier()
+        self._member_key_handler = hs.get_member_key_handler()
         self.enabled = hs.config.experimental.msc4428_enabled
 
     async def on_user_erased(self, user_id: str) -> None:
@@ -54,8 +58,11 @@ class MemberInfoHandler:
         if not room_ids:
             return
 
+        # The stable ID that changed: the user's member key in an MSC4430
+        # member key room, their user ID elsewhere.
+        member_keys = await self._store.get_local_member_keys_for_user(user_id)
         token = await self._store.add_member_info_changes(
-            dict.fromkeys(room_ids, (user_id,))
+            {room_id: (member_keys.get(room_id, user_id),) for room_id in room_ids}
         )
         self._notifier.on_new_event(StreamKeyType.MEMBER_INFO, token, rooms=room_ids)
 
@@ -96,7 +103,10 @@ class MemberInfoHandler:
         """The `member_info` entries for the given stable IDs.
 
         Only stable IDs whose mapping differs from the stable ID itself get an
-        entry; today that is erased users, whose user ID is `None`.
+        entry: a user ID gets one only once the user is erased (`None`); a
+        member key always gets one, naming the user behind it (see
+        `MemberKeyHandler.user_ids_for_identities`), `None` if that user is
+        erased or the key cannot be attributed to anyone.
 
         Args:
             room_id: The room the stable IDs belong to.
@@ -108,12 +118,19 @@ class MemberInfoHandler:
         if not stable_ids:
             return {}
 
-        erased = await self._store.are_users_erased(stable_ids)
-        return {
-            stable_id: {MemberInfo.USER_ID: None}
-            for stable_id, is_erased in erased.items()
-            if is_erased
-        }
+        user_ids = await self._member_key_handler.user_ids_for_identities(
+            room_id, stable_ids
+        )
+        erased = await self._store.are_users_erased(
+            {user_id for user_id in user_ids.values() if user_id is not None}
+        )
+        result: JsonDict = {}
+        for stable_id, user_id in user_ids.items():
+            if user_id is not None and erased.get(user_id):
+                user_id = None
+            if user_id != stable_id:
+                result[stable_id] = {MemberInfo.USER_ID: user_id}
+        return result
 
     async def member_info_for_response(
         self,

synapse/handlers/member_keys.py

@@ -37,7 +37,7 @@ vouch for linkage information that is not ours.
 """
 
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Iterable
 
 from prometheus_client import Counter
 from signedjson.key import generate_signing_key
@@ -162,15 +162,14 @@ class MemberKeyHandler:
         behind one of our member keys, or the *verified* link of another
         server's key. `None` for a key we hold no verified link for: such a key
         maps to no user, whatever it claims."""
-        if not is_member_key(identity):
-            return identity
-        local_user_id = await self.local_user_id_for_identity(room_id, identity)
-        if local_user_id is not None:
-            return local_user_id
-        link = await self._store.get_member_link(room_id, identity)
-        if link is None:
-            return None
-        return link.verified_user_id
+        return (await self.user_ids_for_identities(room_id, [identity]))[identity]
+
+    async def user_ids_for_identities(
+        self, room_id: str, identities: Iterable[str]
+    ) -> dict[str, str | None]:
+        """Batched `user_id_for_identity`: every identity given maps to a user ID
+        or to `None`."""
+        return await self._store.get_user_ids_for_identities(room_id, identities)
 
     async def domain_for_identity(self, room_id: str, identity: str) -> str | None:
         """The server a room member identity belongs to, for routing only: the

synapse/handlers/sliding_sync/__init__.py

@@ -107,6 +107,7 @@ class SlidingSyncHandler:
         self.event_sources = hs.get_event_sources()
         self.relations_handler = hs.get_relations_handler()
         self.member_info_handler = hs.get_member_info_handler()
+        self._member_key_handler = hs.get_member_key_handler()
         self.rooms_to_exclude_globally = hs.config.server.rooms_to_exclude_from_sync
         self.is_mine_id = hs.is_mine_id
 
@@ -611,6 +612,11 @@ class SlidingSyncHandler:
             is_dm: Whether the room is a DM room
         """
         user = sync_config.user
+        # The user as this room knows them: their member key, in an MSC4430
+        # member key room. It is what `$ME` and the heroes are about.
+        user_room_identity = await self._member_key_handler.room_identity_for_user(
+            room_id, user.to_string()
+        )
 
         set_tag(
             SynapseTags.FUNC_ARG_PREFIX + "membership",
@@ -1009,7 +1015,7 @@ class SlidingSyncHandler:
                 # TODO: Reverse/rewind back to the `to_token`
 
             hero_user_ids = extract_heroes_from_room_summary(
-                room_membership_summary, me=user.to_string()
+                room_membership_summary, me=user_room_identity
             )
 
         # Fetch the membership counts for rooms we're joined to.
@@ -1180,7 +1186,7 @@ class SlidingSyncHandler:
                             # their user ID.
                             normalized_state_key = state_key
                             if state_key == StateValues.ME:
-                                normalized_state_key = user.to_string()
+                                normalized_state_key = user_room_identity
 
                             if state_type == EventTypes.Member:
                                 # Also track explicitly requested member state for
@@ -1320,7 +1326,7 @@ class SlidingSyncHandler:
                 # Check if there are any changes to the required state config
                 # that we need to handle.
                 changes_return = _required_state_changes(
-                    user.to_string(),
+                    user_room_identity,
                     prev_required_state_map=prev_room_sync_config.required_state_map,
                     request_required_state_map=room_sync_config.required_state_map,
                     previously_returned_lazy_user_ids=previously_returned_user_to_last_seen.keys(),

synapse/handlers/sliding_sync/room_lists.py

@@ -45,7 +45,7 @@ from synapse.storage.databases.main.state import (
     Sentinel as StateSentinel,
 )
 from synapse.storage.databases.main.stream import CurrentStateDeltaMembership
-from synapse.storage.invite_rule import InviteRule
+from synapse.storage.invite_rule import InviteRule, InviteRulesConfig
 from synapse.storage.roommember import (
     RoomsForUser,
     RoomsForUserSlidingSync,
@@ -189,6 +189,7 @@ def filter_membership_for_sync(
     user_id: str,
     room_membership_for_user: RoomsForUserType,
     newly_left: bool,
+    member_keys_by_room: Mapping[str, str] | None = None,
 ) -> bool:
     """
     Returns True if the membership event should be included in the sync response,
@@ -214,13 +215,20 @@ def filter_membership_for_sync(
     # from the room without a corresponding leave event. We can just remove the rooms
     # since they are no longer relevant to the user but will still appear if they are
     # `newly_left`.
+    # The user as the room knows them: in an MSC4430 member key room, their
+    # member key there (`member_keys_by_room`, from
+    # `get_local_member_keys_for_user`), under which their own leave is sent.
+    own_identities: tuple[str | None, ...] = (user_id, None)
+    if member_keys_by_room:
+        own_identities += (member_keys_by_room.get(room_membership_for_user.room_id),)
+
     return (
         # Anything except leave events
         membership != Membership.LEAVE
         # Unless...
         or newly_left
         # Allow kicks
-        or (membership == Membership.LEAVE and sender not in (user_id, None))
+        or (membership == Membership.LEAVE and sender not in own_identities)
     )
 
 
@@ -234,6 +242,40 @@ class SlidingSyncRoomLists:
         self.is_mine_id = hs.is_mine_id
         self._clock = hs.get_clock()
         self._member_info_handler = hs.get_member_info_handler()
+        self._msc4430_enabled = hs.config.experimental.msc4430_enabled
+
+    async def _invite_is_ignored(
+        self,
+        room_id: str,
+        inviter: str,
+        ignored_users: AbstractSet[str],
+        invite_config: InviteRulesConfig | None,
+    ) -> bool:
+        """Whether an invite is hidden because of who sent it: the inviter is on
+        the user's ignore list, or (when given) the user's invite rules ignore
+        them. In an MSC4430 member key room the inviter is a member key, which
+        may itself be ignored, and which maps to a user (if to anyone) for the
+        user-keyed checks; a key that maps to nobody is not ignored by them."""
+        if inviter in ignored_users:
+            return True
+        inviter_user_id = (
+            await self.store.get_user_ids_for_identities(room_id, [inviter])
+        )[inviter]
+        if inviter_user_id is None:
+            return False
+        if inviter_user_id in ignored_users:
+            return True
+        return (
+            invite_config is not None
+            and invite_config.get_invite_rule(inviter_user_id) == InviteRule.IGNORE
+        )
+
+    async def _member_keys_by_room(self, user_id: str) -> Mapping[str, str]:
+        """The user's member keys per MSC4430 member key room: the identities
+        their own membership events carry there. Empty with the feature off."""
+        if not self._msc4430_enabled:
+            return {}
+        return await self.store.get_local_member_keys_for_user(user_id)
 
     async def compute_interested_rooms(
         self,
@@ -281,6 +323,7 @@ class SlidingSyncRoomLists:
     ) -> SlidingSyncInterestedRooms:
         """Implementation of `compute_interested_rooms` using new sliding sync db tables."""
         user_id = sync_config.user.to_string()
+        member_keys_by_room = await self._member_keys_by_room(user_id)
 
         # Assemble sliding window lists
         lists: dict[str, SlidingSyncResult.SlidingWindowList] = {}
@@ -331,12 +374,11 @@ class SlidingSyncRoomLists:
                 if (
                     room_for_user_sliding_sync.membership == Membership.INVITE
                     and room_for_user_sliding_sync.sender
-                    and (
-                        room_for_user_sliding_sync.sender in ignored_users
-                        or invite_config.get_invite_rule(
-                            room_for_user_sliding_sync.sender
-                        )
-                        == InviteRule.IGNORE
+                    and await self._invite_is_ignored(
+                        room_id,
+                        room_for_user_sliding_sync.sender,
+                        ignored_users,
+                        invite_config,
                     )
                 ):
                     room_membership_for_user_map.pop(room_id, None)
@@ -377,6 +419,7 @@ class SlidingSyncRoomLists:
                         user_id=user_id,
                         room_membership_for_user=room_for_user,
                         newly_left=room_id in newly_left_room_map,
+                        member_keys_by_room=member_keys_by_room,
                     ):
                         room_membership_for_user_map[room_id] = room_for_user
                     else:
@@ -412,6 +455,7 @@ class SlidingSyncRoomLists:
                         user_id=user_id,
                         room_membership_for_user=newly_left_room_for_user_sliding_sync,
                         newly_left=room_id in newly_left_room_map,
+                        member_keys_by_room=member_keys_by_room,
                     ):
                         room_membership_for_user_map[room_id] = (
                             newly_left_room_for_user_sliding_sync
@@ -438,6 +482,7 @@ class SlidingSyncRoomLists:
                             user_id=user_id,
                             room_membership_for_user=room_for_user,
                             newly_left=room_id in newly_left_room_map,
+                            member_keys_by_room=member_keys_by_room,
                         ):
                             room_membership_for_user_map[room_id] = room_for_user
                         else:
@@ -478,6 +523,7 @@ class SlidingSyncRoomLists:
                         user_id=user_id,
                         room_membership_for_user=room_for_user,
                         newly_left=room_id in newly_left_room_map,
+                        member_keys_by_room=member_keys_by_room,
                     ):
                         room_membership_for_user_map[room_id] = room_for_user
                     else:
@@ -1200,14 +1246,20 @@ class SlidingSyncRoomLists:
         # Remove invites from ignored users
         ignored_users = await self.store.ignored_users(user_id)
         if ignored_users:
-            room_for_user_list = [
-                room_for_user
-                for room_for_user in room_for_user_list
-                if not (
+            kept_room_for_user_list = []
+            for room_for_user in room_for_user_list:
+                if (
                     room_for_user.membership == Membership.INVITE
-                    and room_for_user.sender in ignored_users
-                )
-            ]
+                    and await self._invite_is_ignored(
+                        room_for_user.room_id,
+                        room_for_user.sender,
+                        ignored_users,
+                        None,
+                    )
+                ):
+                    continue
+                kept_room_for_user_list.append(room_for_user)
+            room_for_user_list = kept_room_for_user_list
 
         (
             newly_joined_room_ids,
@@ -1502,6 +1554,7 @@ class SlidingSyncRoomLists:
             with membership information in that room at the time of `to_token`.
         """
         user_id = user.to_string()
+        member_keys_by_room = await self._member_keys_by_room(user_id)
 
         # Filter rooms to only what we're interested to sync with
         filtered_sync_room_map = {
@@ -1511,6 +1564,7 @@ class SlidingSyncRoomLists:
                 user_id=user_id,
                 room_membership_for_user=room_membership_for_user,
                 newly_left=room_id in newly_left_room_ids,
+                member_keys_by_room=member_keys_by_room,
             )
         }
 

synapse/handlers/sync.py

@@ -64,7 +64,7 @@ from synapse.metrics import SERVER_NAME_LABEL
 from synapse.storage.databases.main.event_push_actions import RoomNotifCounts
 from synapse.storage.databases.main.roommember import extract_heroes_from_room_summary
 from synapse.storage.databases.main.stream import PaginateFunction
-from synapse.storage.invite_rule import InviteRule
+from synapse.storage.invite_rule import InviteRule, InviteRulesConfig
 from synapse.storage.roommember import MemberSummary
 from synapse.types import (
     DeviceListUpdates,
@@ -429,6 +429,8 @@ class SyncHandler:
         self._storage_controllers = hs.get_storage_controllers()
         self._state_storage_controller = self._storage_controllers.state
         self._member_info_handler = hs.get_member_info_handler()
+        self._member_key_handler = hs.get_member_key_handler()
+        self._msc4430_enabled = hs.config.experimental.msc4430_enabled
         self._device_handler = hs.get_device_handler()
         self._task_scheduler = hs.get_task_scheduler()
 
@@ -1163,7 +1165,10 @@ class SyncHandler:
             for user_id, event_id in details.get(membership, empty_ms).members:
                 member_ids[user_id] = event_id
 
-        me = sync_config.user.to_string()
+        # The user as the room knows them (their member key, in an MSC4430 room).
+        me = await self._member_key_handler.room_identity_for_user(
+            room_id, sync_config.user.to_string()
+        )
         summary["m.heroes"] = extract_heroes_from_room_summary(details, me)
 
         if not sync_config.filter_collection.lazy_load_members():
@@ -1500,7 +1505,15 @@ class SyncHandler:
             # is no guarantee that our membership will be in the auth events of
             # timeline events when the room is partial stated.
             state_filter = StateFilter.from_lazy_load_member_list(
-                members_to_fetch.union((sync_config.user.to_string(),))
+                members_to_fetch.union(
+                    (
+                        # Under the identity the room knows us by (our member
+                        # key, in an MSC4430 room).
+                        await self._member_key_handler.room_identity_for_user(
+                            room_id, sync_config.user.to_string()
+                        ),
+                    )
+                )
             )
 
             # We are happy to use partial state to compute the `/sync` response.
@@ -2060,6 +2073,16 @@ class SyncHandler:
                     newly_joined_or_invited_or_knocked_users,
                     newly_left_users,
                 ) = sync_result_builder.calculate_user_changes()
+                # Those are membership state keys: in an MSC4430 member key
+                # room, keys. Presence and device lists are about users.
+                newly_joined_or_invited_or_knocked_users = (
+                    await self._users_for_member_identities(
+                        sync_result_builder, newly_joined_or_invited_or_knocked_users
+                    )
+                )
+                newly_left_users = await self._users_for_member_identities(
+                    sync_result_builder, newly_left_users
+                )
 
                 if include_presence_data:
                     logger.debug("Fetching presence data")
@@ -2217,8 +2240,15 @@ class SyncHandler:
                 # User joined a room - we have to then check the room state to ensure we
                 # respect any bans if there's a race between the join and ban events.
                 if event.membership == Membership.JOIN:
-                    user_ids_in_room = await self.store.get_users_in_room(room_id)
-                    if user_id in user_ids_in_room:
+                    # (Keyed by user ID, which in an MSC4430 member key room
+                    # the room's own state is not.)
+                    (
+                        current_membership,
+                        _,
+                    ) = await self.store.get_local_current_membership_for_user_in_room(
+                        user_id, room_id
+                    )
+                    if current_membership == Membership.JOIN:
                         mutable_joined_room_ids.add(room_id)
                 # The user left the room, or left and was re-invited but not joined yet
                 else:
@@ -2736,7 +2766,14 @@ class SyncHandler:
         extra_users_ids = set(newly_joined_or_invited_users)
         for room_id in newly_joined_rooms:
             users = await self.store.get_users_in_room(room_id)
-            extra_users_ids.update(users)
+            # (Member keys in an MSC4430 room; presence is about users.)
+            extra_users_ids.update(
+                u
+                for u in (
+                    await self.store.get_user_ids_for_identities(room_id, users)
+                ).values()
+                if u is not None
+            )
         extra_users_ids.discard(user.to_string())
 
         if extra_users_ids:
@@ -2966,6 +3003,67 @@ class SyncHandler:
                 return True
         return False
 
+    async def _users_for_member_identities(
+        self,
+        sync_result_builder: "SyncResultBuilder",
+        identities: AbstractSet[str],
+    ) -> AbstractSet[str]:
+        """Map membership state keys found in the joined rooms' timelines (see
+        `SyncResultBuilder.calculate_user_changes`) to users: user IDs are kept,
+        and in MSC4430 member key rooms the keys become the users behind them
+        (dropped if that is nobody)."""
+        if not identities or not self._msc4430_enabled:
+            return identities
+        # Which room each key was seen in, so it can be looked up there.
+        identity_rooms: dict[str, str] = {}
+        for joined_sync in sync_result_builder.joined:
+            for ev in joined_sync.timeline.events:
+                if (
+                    ev.event.type == EventTypes.Member
+                    and ev.event.state_key in identities
+                ):
+                    identity_rooms[ev.event.state_key] = joined_sync.room_id
+        users: set[str] = set()
+        by_room: dict[str, set[str]] = {}
+        for identity in identities:
+            room_id = identity_rooms.get(identity)
+            if room_id is None:
+                users.add(identity)
+            else:
+                by_room.setdefault(room_id, set()).add(identity)
+        for room_id, room_identities in by_room.items():
+            mapped = await self.store.get_user_ids_for_identities(
+                room_id, room_identities
+            )
+            users.update(u for u in mapped.values() if u is not None)
+        return users
+
+    async def _invite_is_ignored(
+        self,
+        room_id: str,
+        inviter: str,
+        ignored_users: AbstractSet[str],
+        invite_config: InviteRulesConfig,
+    ) -> bool:
+        """Whether an invite is hidden from the user because of who sent it: the
+        inviter is on their ignore list, or their invite rules ignore them. In
+        an MSC4430 member key room the inviter is a member key, which the user
+        may have ignored as such, and which maps to a user (if it maps to any)
+        for the user-keyed checks; a key that maps to nobody cannot be matched
+        by the rules and is not ignored by them.
+        """
+        if inviter in ignored_users:
+            return True
+        inviter_user_id = await self._member_key_handler.user_id_for_identity(
+            room_id, inviter
+        )
+        if inviter_user_id is None:
+            return False
+        return (
+            inviter_user_id in ignored_users
+            or invite_config.get_invite_rule(inviter_user_id) == InviteRule.IGNORE
+        )
+
     async def _get_room_changes_for_incremental_sync(
         self,
         sync_result_builder: "SyncResultBuilder",
@@ -2994,6 +3092,13 @@ class SyncHandler:
         """
         user_id = sync_result_builder.sync_config.user.to_string()
         since_token = sync_result_builder.since_token
+        # The user's identities in MSC4430 member key rooms (their keys), under
+        # which their membership is held in those rooms' state.
+        member_keys_by_room: Mapping[str, str] = (
+            await self.store.get_local_member_keys_for_user(user_id)
+            if self._msc4430_enabled
+            else {}
+        )
         now_token = sync_result_builder.now_token
         sync_config = sync_result_builder.sync_config
         membership_change_events = sync_result_builder.membership_change_events
@@ -3040,13 +3145,18 @@ class SyncHandler:
                 # User is in the room so we don't need to do the invite/leave checks
                 continue
 
+            room_user_id = member_keys_by_room.get(room_id, user_id)
             if room_id in sync_result_builder.joined_room_ids or has_join:
                 old_state_ids = await self._state_storage_controller.get_state_ids_at(
                     room_id,
                     since_token,
-                    state_filter=StateFilter.from_types([(EventTypes.Member, user_id)]),
+                    state_filter=StateFilter.from_types(
+                        [(EventTypes.Member, room_user_id)]
+                    ),
+                )
+                old_mem_ev_id = old_state_ids.get(
+                    (EventTypes.Member, room_user_id), None
                 )
-                old_mem_ev_id = old_state_ids.get((EventTypes.Member, user_id), None)
                 old_mem_ev = None
                 if old_mem_ev_id:
                     old_mem_ev = await self.store.get_event(
@@ -3076,12 +3186,12 @@ class SyncHandler:
                                 room_id,
                                 since_token,
                                 state_filter=StateFilter.from_types(
-                                    [(EventTypes.Member, user_id)]
+                                    [(EventTypes.Member, room_user_id)]
                                 ),
                             )
                         )
                         old_mem_ev_id = old_state_ids.get(
-                            (EventTypes.Member, user_id), None
+                            (EventTypes.Member, room_user_id), None
                         )
                         old_mem_ev = None
                         if old_mem_ev_id:
@@ -3094,10 +3204,8 @@ class SyncHandler:
             # Only bother if we're still currently invited
             should_invite = last_non_join.membership == Membership.INVITE
             if should_invite:
-                if (
-                    last_non_join.sender not in ignored_users
-                    and invite_config.get_invite_rule(last_non_join.sender)
-                    != InviteRule.IGNORE
+                if not await self._invite_is_ignored(
+                    room_id, last_non_join.sender, ignored_users, invite_config
                 ):
                     invite_room_sync = InvitedSyncResult(room_id, invite=last_non_join)
                     if invite_room_sync:
@@ -3255,6 +3363,14 @@ class SyncHandler:
         )
         invite_config = await self.store.get_invite_config_for_user(user_id)
 
+        # The user's identities in MSC4430 member key rooms (their keys), to tell
+        # their own leaves from kicks below.
+        member_keys_by_room: Mapping[str, str] = (
+            await self.store.get_local_member_keys_for_user(user_id)
+            if self._msc4430_enabled
+            else {}
+        )
+
         room_entries = []
         invited = []
         knocked = []
@@ -3277,9 +3393,9 @@ class SyncHandler:
                     )
                 )
             elif event.membership == Membership.INVITE:
-                if event.sender in ignored_users:
-                    continue
-                if invite_config.get_invite_rule(event.sender) == InviteRule.IGNORE:
+                if await self._invite_is_ignored(
+                    event.room_id, event.sender, ignored_users, invite_config
+                ):
                     continue
                 invite = await self.store.get_event(event.event_id)
                 invited.append(InvitedSyncResult(room_id=event.room_id, invite=invite))
@@ -3290,7 +3406,10 @@ class SyncHandler:
                 # Always send down rooms we were banned from or kicked from.
                 if not sync_config.filter_collection.include_leave:
                     if event.membership == Membership.LEAVE:
-                        if user_id == event.sender:
+                        if event.sender in (
+                            user_id,
+                            member_keys_by_room.get(event.room_id),
+                        ):
                             continue
 
                 leave_token = now_token.copy_and_replace(

synapse/push/bulk_push_rule_evaluator.py

@@ -43,6 +43,7 @@ from synapse.api.constants import (
 from synapse.api.room_versions import PushRuleRoomFlag
 from synapse.event_auth import auth_types_for_event, get_user_power_level
 from synapse.events import EventBase, relation_from_event
+from synapse.events.py_protocol import supports_msc4430_member_keys
 from synapse.events.snapshot import EventContext, EventPersistencePair
 from synapse.logging.context import make_deferred_yieldable, run_in_background
 from synapse.metrics import SERVER_NAME_LABEL
@@ -130,6 +131,7 @@ class BulkPushRuleEvaluator:
 
     def __init__(self, hs: "HomeServer"):
         self.hs = hs
+        self._member_key_handler = hs.get_member_key_handler()
         self.server_name = hs.hostname
         self.store = hs.get_datastores().main
         self.server_name = hs.hostname  # nb must be called this for @measure_func
@@ -169,14 +171,20 @@ class BulkPushRuleEvaluator:
             # We never notify a user about their own actions. This is enforced in
             # `_action_for_event_by_user` in the loop over `rules_by_user`, but we
             # do the same check here to avoid unnecessary DB queries.
-            if event.sender != event.state_key and self.hs.is_mine_id(event.state_key):
+            #
+            # (The target as a user: in an MSC4430 member key room the state
+            # key is their member key.)
+            target_user_id = await self._member_key_handler.local_user_id_for_identity(
+                event.room_id, event.state_key
+            )
+            if event.sender != event.state_key and target_user_id is not None:
                 # Check the target is in the room, to avoid notifying them of
                 # e.g. a pre-emptive ban.
                 target_already_in_room = await self.store.check_local_user_in_room(
-                    event.state_key, event.room_id
+                    target_user_id, event.room_id
                 )
                 if target_already_in_room:
-                    local_users = [event.state_key]
+                    local_users = [target_user_id]
         else:
             # We get the users who may need to be notified by first fetching the
             # local users currently in the room, finding those that have push rules,
@@ -202,13 +210,25 @@ class BulkPushRuleEvaluator:
             and event.type == EventTypes.Member
             and event.membership == Membership.INVITE
         ):
-            invited = event.state_key
-            invite_config = await self.store.get_invite_config_for_user(invited)
-            if invite_config.get_invite_rule(event.sender) != InviteRule.ALLOW:
-                # Invite was blocked or ignored, never notify.
-                return {}
-            if self.hs.is_mine_id(invited) and invited not in local_users:
-                local_users.append(invited)
+            # The invitee and the inviter as users (in an MSC4430 member key
+            # room, member keys; an inviter that maps to nobody cannot be
+            # matched by the invitee's rules).
+            invited = await self._member_key_handler.local_user_id_for_identity(
+                event.room_id, event.state_key
+            )
+            if invited is not None:
+                invite_config = await self.store.get_invite_config_for_user(invited)
+                inviter = await self._member_key_handler.user_id_for_identity(
+                    event.room_id, event.sender
+                )
+                if (
+                    inviter is not None
+                    and invite_config.get_invite_rule(inviter) != InviteRule.ALLOW
+                ):
+                    # Invite was blocked or ignored, never notify.
+                    return {}
+                if invited not in local_users:
+                    local_users.append(invited)
 
         if not local_users:
             return {}
@@ -484,8 +504,18 @@ class BulkPushRuleEvaluator:
                 event.room_id, thread_id
             )
 
+        # The sender as a user (their member key maps to them in an MSC4430
+        # member key room): we never notify a user about their own events.
+        sender_user_id = (
+            await self._member_key_handler.local_user_id_for_identity(
+                event.room_id, event.sender
+            )
+            if supports_msc4430_member_keys(event)
+            else event.sender
+        )
+
         for uid, rules in rules_by_user.items():
-            if event.sender == uid:
+            if uid in (event.sender, sender_user_id):
                 continue
 
             display_name = None

synapse/storage/databases/main/member_keys.py

@@ -234,6 +234,44 @@ class MemberKeysWorkerStore(CacheInvalidationWorkerStore):
                     local_users[member_key] = local_key.user_id
         return local_users
 
+    async def get_user_ids_for_identities(
+        self, room_id: str, identities: Iterable[str]
+    ) -> dict[str, str | None]:
+        """The user behind each room member identity: a user ID is its own; a
+        member key we hold in the room maps to our user; any other member key
+        maps to the user its link claims if that link is *verified*, else to
+        `None` (an unverified claim is anyone's to make, so such a key is
+        nobody's). Member keys map to `None` when MSC4430 is off.
+
+        Rooms without member keys query nothing.
+        """
+        users: dict[str, str | None] = {}
+        member_keys: list[str] = []
+        for identity in identities:
+            if is_member_key(identity):
+                users[identity] = None
+                if self._msc4430_enabled:
+                    member_keys.append(identity)
+            else:
+                users[identity] = identity
+        if not member_keys:
+            return users
+
+        local_keys = await self.get_local_member_keys_by_key(room_id, member_keys)
+        remote_keys = []
+        for member_key in member_keys:
+            local_key = local_keys.get(member_key)
+            if local_key is not None:
+                users[member_key] = local_key.user_id
+            else:
+                remote_keys.append(member_key)
+        if remote_keys:
+            links = await self.get_member_links(room_id, remote_keys)
+            for member_key, link in links.items():
+                if link is not None:
+                    users[member_key] = link.verified_user_id
+        return users
+
     async def get_hosts_for_identities(
         self, room_id: str, identities: Iterable[str]
     ) -> dict[str, str]:

synapse/storage/databases/main/stream.py

@@ -942,6 +942,36 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
         return ret, next_key, limited
 
     @trace
+    async def _membership_state_key_clause(
+        self, table_alias: str, user_id: str, column: str = "state_key"
+    ) -> tuple[str, list[str]]:
+        """An SQL clause matching the given user's membership rows in a table
+        whose `column` holds the membership state key as the room knows it.
+
+        In rooms without member keys that is the user ID. In an MSC4430 member
+        key room it is the user's member key for that room, so with MSC4430 on
+        the clause also matches the keys we hold for the user, each scoped to
+        its room (keys are public strings anyone can name as a state key: a key
+        of ours from another room is not this user). The keys come from the
+        cached per-user lookup and are spelled out, so every arm of the clause
+        can use the table's `(room_id, column)` or `column` index; a user with
+        no keys, and the feature being off, give the legacy clause byte for
+        byte.
+
+        The table must have a `room_id` column.
+        """
+        if not self._msc4430_enabled:
+            return f"{table_alias}.{column} = ?", [user_id]
+        member_keys_by_room = await self.get_local_member_keys_for_user(user_id)  # type: ignore[attr-defined]
+        if not member_keys_by_room:
+            return f"{table_alias}.{column} = ?", [user_id]
+        arms = [f"{table_alias}.{column} = ?"]
+        args = [user_id]
+        for room_id, member_key in member_keys_by_room.items():
+            arms.append(f"({table_alias}.room_id = ? AND {table_alias}.{column} = ?)")
+            args.extend((room_id, member_key))
+        return "(" + " OR ".join(arms) + ")", args
+
     async def get_current_state_delta_membership_changes_for_user(
         self,
         user_id: str,
@@ -1002,13 +1032,20 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             if not has_changed:
                 return []
 
+        # In an MSC4430 member key room the user's membership is under their
+        # member key, see `_membership_state_key_clause`.
+        state_key_clause, state_key_args = await self._membership_state_key_clause(
+            "s", user_id
+        )
+
         def f(txn: LoggingTransaction) -> list[CurrentStateDeltaMembership]:
             # To handle tokens with a non-empty instance_map we fetch more
             # results than necessary and then filter down
             min_from_id = from_key.stream
             max_to_id = to_key.get_max_stream_pos()
 
-            args: list[Any] = [min_from_id, max_to_id, EventTypes.Member, user_id]
+            args: list[Any] = [min_from_id, max_to_id, EventTypes.Member]
+            args.extend(state_key_args)
 
             # TODO: It would be good to assert that the `from_token`/`to_token` is >=
             # the first row in `current_state_delta_stream` for the rooms we're
@@ -1045,9 +1082,9 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
                     LEFT JOIN room_memberships AS m_prev ON m_prev.event_id = s.prev_event_id
                 WHERE s.stream_id > ? AND s.stream_id <= ?
                     AND s.type = ?
-                    AND s.state_key = ?
+                    AND %s
                 ORDER BY s.stream_id ASC
-            """
+            """ % (state_key_clause,)
 
             txn.execute(sql, args)
 
@@ -1192,6 +1229,10 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
         if excluded_room_ids is not None:
             room_ids_to_exclude = excluded_room_ids
 
+        state_key_clause, state_key_args = await self._membership_state_key_clause(
+            "s", user_id
+        )
+
         def f(txn: LoggingTransaction) -> dict[str, RoomsForUserStateReset]:
             # To handle tokens with a non-empty instance_map we fetch more
             # results than necessary and then filter down
@@ -1201,7 +1242,10 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             # This query looks at membership changes in
             # `sliding_sync_membership_snapshots` which will not include users
             # that were state reset out of rooms; so we need to look for that
-            # case in `current_state_delta_stream`.
+            # case in `current_state_delta_stream`. (The snapshots are keyed by
+            # user ID in every room; the state deltas by the room's state key,
+            # which in an MSC4430 member key room is the user's member key, see
+            # `_membership_state_key_clause`.)
             sql = """
                 SELECT
                     room_id,
@@ -1243,16 +1287,16 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
                         LEFT JOIN room_memberships AS m_prev ON m_prev.event_id = s.prev_event_id
                     WHERE
                         s.type = ?
-                        AND s.state_key = ?
+                        AND %s
                 ) AS c
                 INNER JOIN rooms USING (room_id)
                 WHERE event_stream_ordering > ? AND event_stream_ordering <= ?
                 ORDER BY event_stream_ordering ASC
-            """
+            """ % (state_key_clause,)
 
             txn.execute(
                 sql,
-                (user_id, EventTypes.Member, user_id, min_from_id, max_to_id),
+                (user_id, EventTypes.Member, *state_key_args, min_from_id, max_to_id),
             )
 
             membership_changes: dict[str, RoomsForUserStateReset] = {}
@@ -1358,30 +1402,20 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             if not has_changed:
                 return []
 
+        # In an MSC4430 member key room, `room_memberships.user_id` is the
+        # member's state key, i.e. the user's member key for that room, see
+        # `_membership_state_key_clause`.
+        user_clause, user_args = await self._membership_state_key_clause(
+            "m", user_id, column="user_id"
+        )
+
         def f(txn: LoggingTransaction) -> list[_EventDictReturn]:
             # To handle tokens with a non-empty instance_map we fetch more
             # results than necessary and then filter down
             min_from_id = from_key.stream
             max_to_id = to_key.get_max_stream_pos()
 
-            if self._msc4430_enabled:
-                # In an MSC4430 member key room, `room_memberships.user_id` is the
-                # member's state key, i.e. the user's member key for that room,
-                # so the user's membership events there are the ones under the
-                # keys we hold for them. Scoped to the event's room: a key of
-                # ours from another room appearing as a state key here is not
-                # this user (keys are public strings anyone can name).
-                user_clause = """(
-                        m.user_id = ?
-                        OR m.user_id IN (
-                            SELECT k.member_key FROM msc4430_member_keys AS k
-                            WHERE k.room_id = m.room_id AND k.user_id = ?
-                        )
-                    )"""
-                args: list[Any] = [user_id, user_id, min_from_id, max_to_id]
-            else:
-                user_clause = "m.user_id = ?"
-                args = [user_id, min_from_id, max_to_id]
+            args: list[Any] = [*user_args, min_from_id, max_to_id]
 
             ignore_room_clause = ""
             if excluded_rooms is not None and len(excluded_rooms) > 0:

synapse/synapse_rust/events.pyi

@@ -372,6 +372,7 @@ class SerializeEventConfig:
         include_admin_metadata: bool,
         msc4354_enabled: bool,
         msc4428_enabled: bool,
+        requester_member_keys: list[str] | None = None,
     ) -> None: ...
     @property
     def as_client_event(self) -> bool:
@@ -418,6 +419,13 @@ class SerializeEventConfig:
         When ``True``, the sender is repeated in ``unsigned`` as
         ``org.matrix.msc4428.stable_id``."""
 
+    @property
+    def requester_member_keys(self) -> set[str] | None:
+        """MSC4430: the requester's member keys. In a member key room the
+        ``sender`` of the requester's own events is their key in that room, so
+        sender-only fields (``transaction_id``, ``delay_id``) are gated on these
+        as well as on ``requester.user_id``."""
+
 def serialize_events(
     events: list[tuple[Event, str | None]],
     time_now_ms: int,

synapse/visibility.py

@@ -24,6 +24,7 @@ from enum import Enum, auto
 from typing import (
     Collection,
     Final,
+    Mapping,
     Sequence,
 )
 
@@ -36,6 +37,7 @@ from synapse.api.constants import (
     Membership,
 )
 from synapse.events import EventBase
+from synapse.events.py_protocol import supports_msc4430_member_keys
 from synapse.events.snapshot import EventContext
 from synapse.events.utils import FilteredEvent, prune_event
 from synapse.logging.opentracing import trace
@@ -151,6 +153,31 @@ async def filter_and_transform_events_for_client(
             )
 
     types = [_HISTORY_VIS_KEY, (EventTypes.Member, user_id)]
+
+    # MSC4430: in a member key room the user's membership is under their member
+    # key in that room, and the senders are keys which the checks below (ignored
+    # senders, erased senders) must see as the users behind them.
+    member_key_room_ids = {e.room_id for e in events if supports_msc4430_member_keys(e)}
+    user_room_identities: Mapping[str, str] = {}
+    # (room ID, sender) -> the user behind the sender: a key is only a user in
+    # the room it was looked up in.
+    sender_user_ids: dict[tuple[str, str], str | None] = {}
+    if member_key_room_ids:
+        user_room_identities = await storage.main.get_local_member_keys_for_user(
+            user_id
+        )
+        types.extend(
+            (EventTypes.Member, member_key)
+            for room_id, member_key in user_room_identities.items()
+            if room_id in member_key_room_ids
+        )
+        for room_id in member_key_room_ids:
+            mapped = await storage.main.get_user_ids_for_identities(
+                room_id, {e.sender for e in events if e.room_id == room_id}
+            )
+            for sender, sender_user in mapped.items():
+                sender_user_ids[(room_id, sender)] = sender_user
+
     if found_call_invite:
         # We need to fetch the room's join rules state to determine
         # whether to allow call invites in public rooms.
@@ -165,7 +192,16 @@ async def filter_and_transform_events_for_client(
     # Get the users who are ignored by the requesting user.
     ignore_list = await storage.main.ignored_users(user_id)
 
-    erased_senders = await storage.main.are_users_erased(e.sender for e in events)
+    def sender_user_id(event: EventBase) -> str | None:
+        """The user behind the event's sender, for the per-user checks: the
+        sender itself in rooms without member keys."""
+        if event.room_id in member_key_room_ids:
+            return sender_user_ids.get((event.room_id, event.sender))
+        return event.sender
+
+    erased_senders = await storage.main.are_users_erased(
+        {u for u in (sender_user_id(e) for e in events) if u is not None}
+    )
 
     if filter_send_to_client:
         room_ids = {e.room_id for e in events}
@@ -178,17 +214,27 @@ async def filter_and_transform_events_for_client(
 
     def allowed(event: EventBase) -> FilteredEvent | None:
         state_after_event = event_id_to_state.get(event.event_id)
+        # The user as this room knows them: their member key, in a member key
+        # room where they have one.
+        room_user_id = (
+            user_room_identities.get(event.room_id, user_id)
+            if event.room_id in member_key_room_ids
+            else user_id
+        )
+        sender = sender_user_id(event)
         filtered = _check_client_allowed_to_see_event(
-            user_id=user_id,
+            user_id=room_user_id,
             event=event,
             clock=storage.main.clock,
             filter_send_to_client=filter_send_to_client,
-            sender_ignored=event.sender in ignore_list,
+            # A member key may be ignored as such (it may be the only handle a
+            # client has on it) or as the user behind it.
+            sender_ignored=event.sender in ignore_list or sender in ignore_list,
             always_include_ids=always_include_ids,
             retention_policy=retention_policies[event.room_id],
             state=state_after_event,
             is_peeking=is_peeking,
-            sender_erased=erased_senders.get(event.sender, False),
+            sender_erased=erased_senders.get(sender, False) if sender else False,
         )
         if filtered is None:
             return None
@@ -219,10 +265,12 @@ async def filter_and_transform_events_for_client(
         # user's own membership event, so we can just inspect that.
 
         user_membership_event: EventBase | None
-        if event.type == EventTypes.Member and event.state_key == user_id:
+        if event.type == EventTypes.Member and event.state_key == room_user_id:
             user_membership_event = event
         elif state_after_event is not None:
-            user_membership_event = state_after_event.get((EventTypes.Member, user_id))
+            user_membership_event = state_after_event.get(
+                (EventTypes.Member, room_user_id)
+            )
         else:
             # unreachable!
             raise Exception("Missing state for event that is not user's own membership")
@@ -286,7 +334,14 @@ async def filter_event_for_clients_with_state(
     # see events in the room at that point in the DAG, and that shouldn't be decided
     # on those checks.
     if filter_send_to_client:
-        ignored_by = await store.ignored_by(event.sender)
+        ignored_by = set(await store.ignored_by(event.sender))
+        if supports_msc4430_member_keys(event):
+            # A member key may be ignored as such, or as the user behind it.
+            sender_user_id = (
+                await store.get_user_ids_for_identities(event.room_id, [event.sender])
+            )[event.sender]
+            if sender_user_id is not None:
+                ignored_by.update(await store.ignored_by(sender_user_id))
         retention_policy = await store.get_retention_policy_for_room(event.room_id)
 
         for user_id in user_ids:
@@ -301,13 +356,28 @@ async def filter_event_for_clients_with_state(
             ):
                 allowed_user_ids.discard(user_id)
 
+    # MSC4430: in a member key room each user's membership is under their member
+    # key in that room (a user with no key there is under their user ID, which
+    # the state then does not hold).
+    room_identities: dict[str, str] = {u: u for u in allowed_user_ids}
+    if supports_msc4430_member_keys(event):
+        local_member_keys = await store.get_local_member_keys_in_room(event.room_id)
+        member_key_by_user_id = {
+            local_user_id: member_key
+            for member_key, local_user_id in local_member_keys.items()
+        }
+        for user_id in allowed_user_ids:
+            room_identities[user_id] = member_key_by_user_id.get(user_id, user_id)
+
     if event.internal_metadata.outlier:
         # Normally these can't be seen by clients, but we make an exception for
         # for out-of-band membership events (eg, incoming invites, or rejections of
         # said invite) for the user themselves.
-        if event.type == EventTypes.Member and event.state_key in allowed_user_ids:
-            logger.debug("Returning out-of-band-membership event %s", event)
-            return {event.state_key}
+        if event.type == EventTypes.Member:
+            for user_id in allowed_user_ids:
+                if room_identities[user_id] == event.state_key:
+                    logger.debug("Returning out-of-band-membership event %s", event)
+                    return {user_id}
 
         return set()
 
@@ -329,7 +399,7 @@ async def filter_event_for_clients_with_state(
 
     filter_list = []
     for user_id in allowed_user_ids:
-        filter_list.append((EventTypes.Member, user_id))
+        filter_list.append((EventTypes.Member, room_identities[user_id]))
     filter_list.append((EventTypes.RoomHistoryVisibility, ""))
 
     state_filter = StateFilter.from_types(filter_list)
@@ -339,7 +409,9 @@ async def filter_event_for_clients_with_state(
     return {
         user_id
         for user_id in allowed_user_ids
-        if _check_membership(user_id, event, visibility, state_map, is_peeking).allowed
+        if _check_membership(
+            room_identities[user_id], event, visibility, state_map, is_peeking
+        ).allowed
     }
 
 

tests/rest/client/test_member_key_rooms.py

@@ -23,22 +23,38 @@ client is served (`/sync`, `/event`, `/members`) is a later change, so the room
 inspected through `/state` and the store.
 """
 
+import json
+import urllib.parse
 from http import HTTPStatus
 from unittest.mock import Mock
 
 from twisted.internet.testing import MemoryReactor
 
-from synapse.api.constants import EventContentFields, EventTypes, MemberKeys, Membership
+from synapse.api.constants import (
+    EventContentFields,
+    EventTypes,
+    MemberInfo,
+    MemberKeys,
+    Membership,
+)
 from synapse.api.room_versions import RoomVersions
 from synapse.events import EventBase
 from synapse.rest import admin
-from synapse.rest.client import login, room, room_upgrade_rest_servlet
+from synapse.rest.client import (
+    account_data,
+    login,
+    room,
+    room_upgrade_rest_servlet,
+    sync,
+)
 from synapse.server import HomeServer
+from synapse.storage.member_keys import MemberLink
 from synapse.types import JsonDict, is_member_key
 from synapse.util.clock import Clock
 
 from tests import unittest
-from tests.test_utils.member_key_rooms import Member
+from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
+from tests.test_utils.member_key_rooms import Member, MemberKeyRoom
 from tests.unittest import override_config
 
 ROOM_VERSION = RoomVersions.MSC4430v12
@@ -422,3 +438,460 @@ class MemberKeyRoomsTestCase(unittest.HomeserverTestCase):
             room_version=ROOM_VERSION.identifier,
             expect_code=HTTPStatus.BAD_REQUEST,
         )
+
+
+class MemberKeyClientViewTestCase(unittest.HomeserverTestCase):
+    """What a client is served from a member key room: events whose senders are
+    keys with `stable_id` set, a `member_info` map resolving the keys, the
+    requester's own events recognised for the transaction ID echo, visibility
+    decided by the requester's key, and `/members` and `/joined_members` keyed
+    by key. These are the local halves of the Complement tests for the stable
+    ID, `member_info` in `/sync`, the member list endpoints, the transaction ID
+    echo and the incremental sync after a join."""
+
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+        sync.register_servlets,
+        account_data.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        config.setdefault("experimental_features", {}).setdefault(
+            "msc4430_enabled", True
+        )
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.store = hs.get_datastores().main
+        self.alice = self.register_user("alice", "pass")
+        self.alice_tok = self.login("alice", "pass")
+        self.bob = self.register_user("bob", "pass")
+        self.bob_tok = self.login("bob", "pass")
+        self.room_id = self.helper.create_room_as(
+            self.alice,
+            tok=self.alice_tok,
+            room_version=ROOM_VERSION.identifier,
+            extra_content={"preset": "public_chat"},
+        )
+        self.alice_key = self._key(self.alice)
+
+    def _key(self, user_id: str) -> str:
+        local_key = self.get_success(
+            self.store.get_local_member_key(self.room_id, user_id)
+        )
+        assert local_key is not None
+        return local_key.member_key
+
+    def _sync(self, tok: str, since: str | None = None, **params: str) -> JsonDict:
+        query = {**params}
+        if since is not None:
+            query["since"] = since
+        qs = "&".join(f"{k}={v}" for k, v in query.items())
+        channel = self.make_request(
+            "GET", f"/sync?{qs}" if qs else "/sync", access_token=tok
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        return channel.json_body
+
+    def _joined_room(self, body: JsonDict) -> JsonDict:
+        return body["rooms"]["join"][self.room_id]
+
+    def _stranger_join(self, stranger: Member) -> EventBase:
+        """A join by a key this server does not hold, as federation would bring."""
+        builder = self.get_success(
+            MemberKeyRoom.for_existing_room(self.hs, self.room_id)
+        )
+        return self.get_success(builder.send_member(stranger, "join"))
+
+    def test_initial_sync(self) -> None:
+        """Every event names a key as sender and stable ID; the map resolves
+        every member's key to their user ID."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        self.helper.send(self.room_id, body="hello", tok=self.bob_tok)
+
+        room_body = self._joined_room(self._sync(self.alice_tok))
+        events = room_body["timeline"]["events"] + room_body["state"]["events"]
+        self.assertTrue(events)
+        for event in events:
+            self.assertTrue(is_member_key(event["sender"]), event)
+            self.assertEqual(
+                event["unsigned"]["org.matrix.msc4428.stable_id"], event["sender"]
+            )
+            if event["type"] == EventTypes.Member:
+                self.assertTrue(is_member_key(event["state_key"]), event)
+        self.assertEqual(
+            room_body[MemberInfo.FIELD],
+            {
+                self.alice_key: {MemberInfo.USER_ID: self.alice},
+                bob_key: {MemberInfo.USER_ID: self.bob},
+            },
+        )
+        # The user's own membership is served as state under their key.
+        memberships = {
+            e["state_key"]: e["content"]["membership"]
+            for e in events
+            if e["type"] == EventTypes.Member
+        }
+        self.assertEqual(memberships, {self.alice_key: "join", bob_key: "join"})
+
+    def test_incremental_sync_after_join_and_leave(self) -> None:
+        """A room joined after the last sync comes down with its create event,
+        and only the new member's entry; leaving puts it under `leave`."""
+        since = self._sync(self.bob_tok)["next_batch"]
+        self.assertNotIn(
+            self.room_id, self._sync(self.bob_tok).get("rooms", {}).get("join", {})
+        )
+
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        body = self._sync(self.bob_tok, since=since)
+        room_body = self._joined_room(body)
+        types = {e["type"] for e in room_body["timeline"]["events"]} | {
+            e["type"] for e in room_body["state"]["events"]
+        }
+        self.assertIn(EventTypes.Create, types)
+        self.assertEqual(
+            room_body[MemberInfo.FIELD][bob_key], {MemberInfo.USER_ID: self.bob}
+        )
+
+        # Alice, already synced, learns only of Bob.
+        alice_since = self._sync(self.alice_tok)["next_batch"]
+        self.helper.send(self.room_id, body="hi", tok=self.bob_tok)
+        room_body = self._joined_room(self._sync(self.alice_tok, since=alice_since))
+        self.assertEqual(
+            room_body[MemberInfo.FIELD], {bob_key: {MemberInfo.USER_ID: self.bob}}
+        )
+
+        since = body["next_batch"]
+        self.helper.leave(self.room_id, self.bob, tok=self.bob_tok)
+        body = self._sync(self.bob_tok, since=since)
+        self.assertIn(self.room_id, body["rooms"]["leave"])
+        self.assertNotIn(self.room_id, body["rooms"].get("join", {}))
+
+    def test_transaction_id_echoed_to_the_sending_device_only(self) -> None:
+        other_device_tok = self.login("alice", "pass")
+        event_id = self.helper.send(
+            self.room_id, body="hello", tok=self.alice_tok, txn_id="txn1"
+        )["event_id"]
+
+        def find(body: JsonDict) -> JsonDict:
+            for event in self._joined_room(body)["timeline"]["events"]:
+                if event["event_id"] == event_id:
+                    return event
+            raise AssertionError("event not in timeline")
+
+        self.assertEqual(
+            find(self._sync(self.alice_tok))["unsigned"]["transaction_id"], "txn1"
+        )
+        self.assertNotIn(
+            "transaction_id", find(self._sync(other_device_tok))["unsigned"]
+        )
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        self.assertNotIn("transaction_id", find(self._sync(self.bob_tok))["unsigned"])
+
+    def test_members_and_joined_members(self) -> None:
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+
+        channel = self.make_request(
+            "GET", f"/rooms/{self.room_id}/members", access_token=self.alice_tok
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        self.assertEqual(
+            {e["state_key"] for e in channel.json_body["chunk"]},
+            {self.alice_key, bob_key},
+        )
+        self.assertEqual(
+            channel.json_body[MemberInfo.FIELD],
+            {
+                self.alice_key: {MemberInfo.USER_ID: self.alice},
+                bob_key: {MemberInfo.USER_ID: self.bob},
+            },
+        )
+
+        channel = self.make_request(
+            "GET", f"/rooms/{self.room_id}/joined_members", access_token=self.alice_tok
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        joined = channel.json_body["joined"]
+        self.assertEqual(set(joined), {self.alice_key, bob_key})
+        self.assertEqual(joined[bob_key][MemberInfo.USER_ID], self.bob)
+        self.assertEqual(joined[bob_key]["display_name"], "bob")
+
+    def test_event_and_messages_are_visible_to_members_only(self) -> None:
+        """Visibility is decided by the requester's key: with `joined` history,
+        a later joiner sees nothing from before their join."""
+        self.helper.send_state(
+            self.room_id,
+            EventTypes.RoomHistoryVisibility,
+            {"history_visibility": "joined"},
+            tok=self.alice_tok,
+        )
+        before = self.helper.send(self.room_id, body="before", tok=self.alice_tok)[
+            "event_id"
+        ]
+        self.helper.get_event(self.room_id, before, tok=self.alice_tok)
+        self.helper.get_event(
+            self.room_id, before, tok=self.bob_tok, expect_code=HTTPStatus.NOT_FOUND
+        )
+
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        after = self.helper.send(self.room_id, body="after", tok=self.alice_tok)[
+            "event_id"
+        ]
+        self.helper.get_event(
+            self.room_id, before, tok=self.bob_tok, expect_code=HTTPStatus.NOT_FOUND
+        )
+        self.helper.get_event(self.room_id, after, tok=self.bob_tok)
+
+        channel = self.make_request(
+            "GET",
+            f"/rooms/{self.room_id}/messages?dir=b",
+            access_token=self.bob_tok,
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        bodies = [
+            e["content"]["body"]
+            for e in channel.json_body["chunk"]
+            if e["type"] == EventTypes.Message
+        ]
+        self.assertEqual(bodies, ["after"])
+
+    def test_member_info_for_erased_and_unknown_members(self) -> None:
+        """An erased member maps to null, as does a key nobody vouches for; a
+        verified link names the user it claims."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        self.get_success(self.store.mark_user_erased(self.bob))
+
+        stranger, linked = Member(), Member()
+        self._stranger_join(stranger)
+        self._stranger_join(linked)
+        self.get_success(
+            self.store.upsert_member_link(
+                MemberLink(
+                    room_id=self.room_id,
+                    member_key=linked.key,
+                    domain="other.example",
+                    localpart="carol",
+                    verified=True,
+                    event_id="$verified",
+                    key_id="ed25519:x",
+                )
+            )
+        )
+
+        channel = self.make_request(
+            "GET", f"/rooms/{self.room_id}/members", access_token=self.alice_tok
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        self.assertEqual(
+            channel.json_body[MemberInfo.FIELD],
+            {
+                self.alice_key: {MemberInfo.USER_ID: self.alice},
+                bob_key: {MemberInfo.USER_ID: None},
+                stranger.key: {MemberInfo.USER_ID: None},
+                linked.key: {MemberInfo.USER_ID: "@carol:other.example"},
+            },
+        )
+
+    def _ignore(self, tok: str, user_id: str, *ignored: str) -> None:
+        channel = self.make_request(
+            "PUT",
+            f"/user/{user_id}/account_data/m.ignored_user_list",
+            {"ignored_users": {i: {} for i in ignored}},
+            access_token=tok,
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+
+    def test_profile_change_is_not_a_rejoin(self) -> None:
+        """Changing one's own membership (a display name) is an ordinary
+        timeline event on the next sync, not the room sent again as newly
+        joined; and the room's own leave is a leave, not a kick."""
+        since = self._sync(self.alice_tok)["next_batch"]
+        self.helper.send_state(
+            self.room_id,
+            EventTypes.Member,
+            {"membership": Membership.JOIN, "displayname": "Alice!"},
+            tok=self.alice_tok,
+            state_key=self.alice,
+        )
+        room_body = self._joined_room(self._sync(self.alice_tok, since=since))
+        self.assertEqual(
+            [e["type"] for e in room_body["timeline"]["events"]], [EventTypes.Member]
+        )
+        self.assertFalse(room_body["timeline"]["limited"])
+        self.assertEqual(room_body["state"]["events"], [])
+
+    def test_device_lists_and_presence_name_users(self) -> None:
+        """Device list changes and presence are about users, not keys."""
+        since = self._sync(self.alice_tok)["next_batch"]
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        body = self._sync(self.alice_tok, since=since)
+        self.assertEqual(body["device_lists"]["changed"], [self.bob])
+        self.assertEqual({e["sender"] for e in body["presence"]["events"]}, {self.bob})
+
+        since = body["next_batch"]
+        self.helper.leave(self.room_id, self.bob, tok=self.bob_tok)
+        body = self._sync(self.alice_tok, since=since)
+        self.assertEqual(body["device_lists"]["left"], [self.bob])
+
+    def test_erasure_changes_the_keys_entry(self) -> None:
+        """Erasure is delivered against the member's key, the stable ID in
+        these rooms, not against their user ID."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        since = self._sync(self.alice_tok)["next_batch"]
+
+        self.get_success(self.store.mark_user_erased(self.bob))
+        self.get_success(self.hs.get_member_info_handler().on_user_erased(self.bob))
+        room_body = self._joined_room(self._sync(self.alice_tok, since=since))
+        self.assertEqual(
+            room_body[MemberInfo.FIELD], {bob_key: {MemberInfo.USER_ID: None}}
+        )
+
+    def test_ignored_senders_by_user_id_and_by_key(self) -> None:
+        """A member may be ignored as the user behind their key or as the key
+        itself; either way their events and notifications stop."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        alice_key = self.alice_key
+
+        for ignored in (self.alice, alice_key):
+            self._ignore(self.bob_tok, self.bob, ignored)
+            since = self._sync(self.bob_tok)["next_batch"]
+            self.helper.send(self.room_id, body="ignored?", tok=self.alice_tok)
+            body = self._sync(self.bob_tok, since=since)
+            room_body = body.get("rooms", {}).get("join", {}).get(self.room_id, {})
+            self.assertEqual(room_body.get("timeline", {}).get("events", []), [])
+            self.assertEqual(
+                room_body.get("unread_notifications", {}).get("notification_count", 0),
+                0,
+            )
+            self._ignore(self.bob_tok, self.bob)
+
+    def test_invite_from_ignored_user_is_hidden(self) -> None:
+        self._ignore(self.bob_tok, self.bob, self.alice)
+        self.helper.invite(
+            self.room_id, src=self.alice, targ=self.bob, tok=self.alice_tok
+        )
+        body = self._sync(self.bob_tok)
+        self.assertNotIn(self.room_id, body.get("rooms", {}).get("invite", {}))
+
+    def test_own_events_are_not_notified(self) -> None:
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        since = self._sync(self.alice_tok)["next_batch"]
+        self.helper.send(self.room_id, body="mine", tok=self.alice_tok)
+        room_body = self._joined_room(self._sync(self.alice_tok, since=since))
+        self.assertEqual(room_body["unread_notifications"]["notification_count"], 0)
+
+    def test_heroes_exclude_the_requester(self) -> None:
+        """A nameless room is summarised by the other members' keys (the
+        summary is sent to clients that lazy-load members)."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        lazy = urllib.parse.quote(
+            json.dumps({"room": {"state": {"lazy_load_members": True}}})
+        )
+        summary = self._joined_room(self._sync(self.alice_tok, filter=lazy))["summary"]
+        self.assertEqual(summary["m.heroes"], [bob_key])
+        summary = self._joined_room(self._sync(self.bob_tok, filter=lazy))["summary"]
+        self.assertEqual(summary["m.heroes"], [self.alice_key])
+
+    def test_notification_count(self) -> None:
+        """Push evaluation decides who can see the event by their key."""
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        since = self._sync(self.bob_tok)["next_batch"]
+        self.helper.send(self.room_id, body="ping", tok=self.alice_tok)
+        room_body = self._joined_room(self._sync(self.bob_tok, since=since))
+        self.assertEqual(room_body["unread_notifications"]["notification_count"], 1)
+
+
+class MemberKeySlidingSyncTestCase(SlidingSyncBase):
+    """Sliding sync in a member key room: `$ME` is the requester's key, the room
+    list follows joins and leaves, and `member_info` resolves the keys."""
+
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+        sync.register_servlets,
+        account_data.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        config["experimental_features"]["msc4430_enabled"] = True
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        super().prepare(reactor, clock, hs)
+        self.store = hs.get_datastores().main
+        self.alice = self.register_user("alice", "pass")
+        self.alice_tok = self.login("alice", "pass")
+        self.bob = self.register_user("bob", "pass")
+        self.bob_tok = self.login("bob", "pass")
+        self.room_id = self.helper.create_room_as(
+            self.alice,
+            tok=self.alice_tok,
+            room_version=ROOM_VERSION.identifier,
+            extra_content={"preset": "public_chat"},
+        )
+        self.sync_body = {
+            "lists": {
+                "foo-list": {
+                    "ranges": [[0, 1]],
+                    "required_state": [[EventTypes.Member, "$ME"]],
+                    "timeline_limit": 2,
+                }
+            }
+        }
+
+    def _key(self, user_id: str) -> str:
+        local_key = self.get_success(
+            self.store.get_local_member_key(self.room_id, user_id)
+        )
+        assert local_key is not None
+        return local_key.member_key
+
+    def test_invite_from_ignored_user_is_hidden(self) -> None:
+        channel = self.make_request(
+            "PUT",
+            f"/user/{self.bob}/account_data/m.ignored_user_list",
+            {"ignored_users": {self.alice: {}}},
+            access_token=self.bob_tok,
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+        self.helper.invite(
+            self.room_id, src=self.alice, targ=self.bob, tok=self.alice_tok
+        )
+        body, _ = self.do_sync(self.sync_body, tok=self.bob_tok)
+        self.assertNotIn(self.room_id, body["rooms"])
+
+    def test_me_join_and_leave(self) -> None:
+        body, pos = self.do_sync(self.sync_body, tok=self.bob_tok)
+        self.assertNotIn(self.room_id, body["rooms"])
+
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+        bob_key = self._key(self.bob)
+        body, pos = self.do_sync(self.sync_body, since=pos, tok=self.bob_tok)
+        room_body = body["rooms"][self.room_id]
+        self.assertTrue(room_body["initial"])
+        # `$ME` is Bob's key: his own membership, and nobody else's.
+        self.assertEqual(
+            [(e["type"], e["state_key"]) for e in room_body["required_state"]],
+            [(EventTypes.Member, bob_key)],
+        )
+        self.assertEqual(
+            room_body[MemberInfo.FIELD][bob_key], {MemberInfo.USER_ID: self.bob}
+        )
+
+        # Leaving on one's own removes the room from the list (it is not a kick).
+        self.helper.leave(self.room_id, self.bob, tok=self.bob_tok)
+        body, pos = self.do_sync(self.sync_body, since=pos, tok=self.bob_tok)
+        self.assertIn(self.room_id, body["rooms"])  # newly left: delivered once
+        body, pos = self.do_sync(self.sync_body, since=pos, tok=self.bob_tok)
+        self.assertNotIn(self.room_id, body["rooms"])

tests/storage/test_member_key_persistence.py

@@ -268,6 +268,54 @@ class MemberKeyPersistenceTestCase(unittest.HomeserverTestCase):
             self.get_success(self.store.get_rooms_for_user(self.bob)), frozenset()
         )
 
+    def test_state_delta_membership_changes_for_user(self) -> None:
+        """The two queries over `current_state_delta_stream` find the user's
+        changes under their key: a join, a kick by another key, and nothing
+        for a key of ours from another room."""
+        bob_member = self._local_key(self.bob)
+        stranger = Member()
+        self.get_success(self.room.send_member(self.alice_member, "join"))
+        before = self.store.get_room_max_token()
+        self.get_success(self.room.send_member(bob_member, "join"))
+        kick = self.get_success(
+            self.room.send_member(bob_member, "leave", sender=self.alice_member)
+        )
+        self.get_success(
+            self.store.create_local_member_key(
+                OTHER_ROOM, "@carol:test", stranger.signing_key
+            )
+        )
+        self.get_success(self.room.send_member(stranger, "join"))
+        after = self.store.get_room_max_token()
+
+        changes = self.get_success(
+            self.store.get_current_state_delta_membership_changes_for_user(
+                self.bob, before, after
+            )
+        )
+        self.assertEqual(
+            [(c.room_id, c.membership, c.sender) for c in changes],
+            [
+                (self.room_id, Membership.JOIN, bob_member.key),
+                (self.room_id, Membership.LEAVE, self.alice_member.key),
+            ],
+        )
+        sliding = self.get_success(
+            self.store.get_sliding_sync_membership_changes(self.bob, before, after)
+        )
+        self.assertEqual(
+            {room_id: (c.membership, c.event_id) for room_id, c in sliding.items()},
+            {self.room_id: (Membership.LEAVE, kick.event_id)},
+        )
+        self.assertEqual(
+            self.get_success(
+                self.store.get_current_state_delta_membership_changes_for_user(
+                    "@carol:test", before, after
+                )
+            ),
+            [],
+        )
+
     def test_leave_removes_membership(self) -> None:
         """A leave updates the user's row, and when the leaver was our last
         member the server leaves the room (which depends on recognising the

tests/test_utils/member_key_rooms.py

@@ -202,6 +202,25 @@ class MemberKeyRoom:
             self._depth = max(self._depth, event.depth)
         return persisted
 
+    @classmethod
+    async def for_existing_room(
+        cls, hs: HomeServer, room_id: str, creator: Member | None = None
+    ) -> "MemberKeyRoom":
+        """Attach to a member key room that already exists (one created through
+        the client API, say), chaining new events onto its current extremities.
+        `creator` is only a label: the events sent through the room are signed
+        by whichever `Member` they name."""
+        room = cls(hs, creator or Member())
+        room.room_id = room_id
+        store = hs.get_datastores().main
+        prev_events = await store.get_prev_events_for_room(room_id)
+        room._latest_event_id = prev_events[0]
+        state_extremities = list(await store.get_state_dag_extremities(room_id))
+        room._latest_state_event_id = state_extremities[0]
+        _, depth = await store.get_max_depth_of(prev_events)
+        room._depth = depth
+        return room
+
     async def create(self, content: JsonDict | None = None) -> EventBase:
         """Store the `rooms` row and persist the create event, sent by the
         creator's member key. The room ID is the create event's ID."""