A3: Add a stream for MSC4428 member_info changes with no event anchor

Changes to a member's `member_info` entry that no event in the room anchors
(erasure, and MSC4430 link verification) were only delivered when the member
next appeared in a response. Give them a stream so that syncing clients are
woken and told.

* `StreamToken.member_info_key`: a trailing `MultiWriterStreamToken`
  component (`StreamKeyType.MEMBER_INFO`); tokens without it still parse.
* `msc4428_member_info_changes` (delta 95/01) behind a multi-writer id
  generator with no fixed writer list, as for cache invalidations: any
  worker may write (erasure runs wherever deactivation is handled).
  `MemberInfoChangesStream` replicates rows; workers wake room members.
* Retention: rows older than 30 days are pruned hourly on the background
  worker and the prune point recorded; a since token older than that is
  sent the full map per room instead of deltas.
* Producer: `MemberInfoHandler.on_user_erased`, from `deactivate_account`,
  one row per room the user has any membership in.
* Consumers: sync v2 reads the changes for all joined rooms once per sync
  and includes a room with an empty timeline when it has any; sliding sync
  includes such rooms when choosing what to send and resends the full map
  for rooms it last sent at an older position.

The stream exists always, like every stream, but nothing writes to it unless
`msc4428_enabled` is set.

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

synapse/_scripts/synapse_port_db.py

@@ -860,6 +860,12 @@ class Porter:
                     ("presence_stream", "stream_id"),
                 ],
             )
+            await self._setup_sequence(
+                "msc4428_member_info_changes_sequence",
+                [
+                    ("msc4428_member_info_changes", "stream_id"),
+                ],
+            )
             await self._setup_auth_chain_sequence()
             await self._setup_sequence(
                 "application_services_txn_id_seq",

synapse/app/generic_worker.py

@@ -82,6 +82,7 @@ from synapse.storage.databases.main.filtering import FilteringWorkerStore
 from synapse.storage.databases.main.keys import KeyStore
 from synapse.storage.databases.main.lock import LockStore
 from synapse.storage.databases.main.media_repository import MediaRepositoryStore
+from synapse.storage.databases.main.member_info import MemberInfoWorkerStore
 from synapse.storage.databases.main.metrics import ServerMetricsStore
 from synapse.storage.databases.main.monthly_active_users import (
     MonthlyActiveUsersWorkerStore,
@@ -140,6 +141,7 @@ class GenericWorkerStore(
     DirectoryWorkerStore,
     ThreadSubscriptionsWorkerStore,
     StickyEventsWorkerStore,
+    MemberInfoWorkerStore,
     PushRulesWorkerStore,
     ApplicationServiceTransactionWorkerStore,
     ApplicationServiceWorkerStore,

synapse/handlers/deactivate_account.py

@@ -53,6 +53,7 @@ class DeactivateAccountHandler:
         self._room_member_handler = hs.get_room_member_handler()
         self._identity_handler = hs.get_identity_handler()
         self._profile_handler = hs.get_profile_handler()
+        self._member_info_handler = hs.get_member_info_handler()
         self._delayed_events_handler = hs.get_delayed_events_handler()
         self._pusher_pool = hs.get_pusherpool()
         self.user_directory_handler = hs.get_user_directory_handler()
@@ -193,6 +194,9 @@ class DeactivateAccountHandler:
 
             logger.info("Marking %s as erased", user_id)
             await self.store.mark_user_erased(user_id)
+            # MSC4428: tell the members of the user's rooms that the user ID
+            # behind this stable ID is gone.
+            await self._member_info_handler.on_user_erased(user_id)
 
         # Reject all pending invites and knocks for the user, so that the
         # user doesn't show up in the "invited" section of rooms' members list.

synapse/handlers/member_info.py

@@ -18,13 +18,19 @@ 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.
+
+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
+recorded on the member info changes stream (`MemberInfoWorkerStore`) so that
+syncing clients are woken and told.
 """
 
-from typing import TYPE_CHECKING, Collection, Iterable
+from typing import TYPE_CHECKING, Collection, Iterable, Mapping
 
 from synapse.api.constants import EventTypes, MemberInfo
 from synapse.events import EventBase
-from synapse.types import JsonDict
+from synapse.types import JsonDict, StreamKeyType, StreamToken
+from synapse.types.state import StateFilter
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -33,8 +39,46 @@ if TYPE_CHECKING:
 class MemberInfoHandler:
     def __init__(self, hs: "HomeServer"):
         self._store = hs.get_datastores().main
+        self._state_storage_controller = hs.get_storage_controllers().state
+        self._notifier = hs.get_notifier()
         self.enabled = hs.config.experimental.msc4428_enabled
 
+    async def on_user_erased(self, user_id: str) -> None:
+        """Record that the erased user's `member_info` entry changed in every
+        room they have any membership state in, and wake the members of those
+        rooms. Nothing in the room anchors this change, hence the stream."""
+        if not self.enabled:
+            return
+
+        room_ids = await self._store.get_rooms_with_membership_for_local_user(user_id)
+        if not room_ids:
+            return
+
+        token = await self._store.add_member_info_changes(
+            dict.fromkeys(room_ids, (user_id,))
+        )
+        self._notifier.on_new_event(StreamKeyType.MEMBER_INFO, token, rooms=room_ids)
+
+    async def get_changes_in_rooms(
+        self,
+        room_ids: Collection[str],
+        from_token: StreamToken,
+        to_token: StreamToken,
+    ) -> Mapping[str, Collection[str]] | None:
+        """The stable IDs whose entry changed without an event anchor in the
+        given rooms between the two tokens.
+
+        Returns:
+            room ID -> stable IDs, for rooms with changes only; or None if
+            `from_token` is older than the retained changes, in which case the
+            caller must send the full map (`member_info_for_response(full=True)`).
+        """
+        if not self.enabled:
+            return {}
+        return await self._store.get_member_info_changes_in_rooms(
+            room_ids, from_token.member_info_key, to_token.member_info_key
+        )
+
     @staticmethod
     def stable_ids_from_events(events: Iterable[EventBase]) -> set[str]:
         """The stable IDs a client learns about from a set of events: the
@@ -76,14 +120,25 @@ class MemberInfoHandler:
         room_id: str,
         events: Iterable[EventBase],
         heroes: Iterable[str] = (),
+        *,
+        changed: Iterable[str] = (),
+        full: bool = False,
     ) -> JsonDict | None:
         """The `member_info` map for one room of a sync response.
 
         The map covers the stable IDs the response mentions: the senders of the
-        events in it, the targets of its membership events, and the room's
-        heroes. On an initial sync without lazy-loaded members that is every
-        member of the room; a lazy-loading client gets the entries for the
-        members it loads later from `/members`.
+        events in it, the targets of its membership events, the room's heroes,
+        and the `changed` stable IDs from the change stream. On an initial sync
+        without lazy-loaded members that is every member of the room; a
+        lazy-loading client gets the entries for the members it loads later
+        from `/members`.
+
+        Args:
+            changed: stable IDs from `get_changes_in_rooms` for this room.
+            full: also cover every current member of the room. For responses
+                whose deltas cannot be computed (a token older than the change
+                stream's retention, or a sliding sync room the connection has
+                not seen since its last position).
 
         Returns:
             The map, or `None` if MSC4428 is disabled or there is nothing to say
@@ -94,5 +149,16 @@ class MemberInfoHandler:
 
         stable_ids = self.stable_ids_from_events(events)
         stable_ids.update(heroes)
+        stable_ids.update(changed)
+
+        if full:
+            member_state = await self._state_storage_controller.get_current_state_ids(
+                room_id,
+                StateFilter.from_types([(EventTypes.Member, None)]),
+                # A partial-state room is missing remote members; do not block
+                # the sync on the resync for this fallback path.
+                await_full_state=False,
+            )
+            stable_ids.update(state_key for _, state_key in member_state)
 
         return await self.get_member_info(room_id, stable_ids) or None

synapse/handlers/sliding_sync/__init__.py

@@ -15,7 +15,7 @@
 import itertools
 import logging
 from itertools import chain
-from typing import TYPE_CHECKING, AbstractSet, Mapping
+from typing import TYPE_CHECKING, AbstractSet, Collection, Mapping
 
 import attr
 from prometheus_client import Histogram
@@ -314,6 +314,7 @@ class SlidingSyncHandler:
                 newly_joined=room_id in interested_rooms.newly_joined_rooms,
                 newly_left=room_id in interested_rooms.newly_left_rooms,
                 is_dm=room_id in interested_rooms.dm_room_ids,
+                member_info_changes_by_room=interested_rooms.member_info_changes_by_room,
             )
 
             # Filter out empty room results during incremental sync
@@ -587,6 +588,7 @@ class SlidingSyncHandler:
         newly_joined: bool,
         newly_left: bool,
         is_dm: bool,
+        member_info_changes_by_room: Mapping[str, Collection[str]] | None = None,
     ) -> SlidingSyncResult.RoomResult:
         """
         Fetch room data for the sync response.
@@ -598,6 +600,8 @@ class SlidingSyncHandler:
             room_id: The room ID to fetch data for
             room_sync_config: Config for what data we should fetch for a room in the
                 sync response.
+            member_info_changes_by_room: MSC4428, see
+                `SlidingSyncInterestedRooms.member_info_changes_by_room`.
             room_membership_for_user_at_to_token: Membership information for the user
                 in the room at the time of `to_token`.
             from_token: The point in the stream to sync from.
@@ -668,15 +672,27 @@ class SlidingSyncHandler:
         from_bound = None
         initial = True
         ignore_timeline_bound = False
+        # MSC4428: which `member_info` entries to send beyond those the events
+        # in the response mention (see `member_info_for_response`).
+        member_info_changed: Collection[str] = ()
+        member_info_full = False
         if from_token and not newly_joined and not state_reset_out_of_room:
             room_status = previous_connection_state.rooms.have_sent_room(room_id)
             if room_status.status == HaveSentRoomFlag.LIVE:
                 from_bound = from_token.stream_token.room_key
                 initial = False
+                if member_info_changes_by_room is None:
+                    # The token predates the retained changes.
+                    member_info_full = True
+                else:
+                    member_info_changed = member_info_changes_by_room.get(room_id, ())
             elif room_status.status == HaveSentRoomFlag.PREVIOUSLY:
                 assert room_status.last_token is not None
                 from_bound = room_status.last_token
                 initial = False
+                # We only know the room's events-stream position from when it
+                # was last sent, not its member info position: resend the map.
+                member_info_full = True
             elif room_status.status == HaveSentRoomFlag.NEVER:
                 from_bound = None
                 initial = True
@@ -1506,6 +1522,8 @@ class SlidingSyncHandler:
                 (ev.event for ev in timeline_events), required_room_state.values()
             ),
             heroes=(hero.user_id for hero in heroes),
+            changed=member_info_changed,
+            full=member_info_full,
         )
 
         return SlidingSyncResult.RoomResult(

synapse/handlers/sliding_sync/room_lists.py

@@ -18,6 +18,7 @@ from itertools import chain
 from typing import (
     TYPE_CHECKING,
     AbstractSet,
+    Collection,
     Literal,
     Mapping,
     MutableMapping,
@@ -161,6 +162,14 @@ class SlidingSyncInterestedRooms:
     The set of rooms the user consider as direct-message (DM) rooms
     """
 
+    member_info_changes_by_room: Mapping[str, Collection[str]] | None = None
+    """
+    MSC4428: for an incremental sync, the stable IDs whose `member_info` entry
+    changed with no event anchor since `from_token`, per room in
+    `relevant_room_map` (rooms with changes only). None if the token predates
+    the retained changes, in which case every room needs the full map.
+    """
+
     @staticmethod
     def empty() -> "SlidingSyncInterestedRooms":
         return SlidingSyncInterestedRooms(
@@ -224,6 +233,7 @@ class SlidingSyncRoomLists:
         self.rooms_to_exclude_globally = hs.config.server.rooms_to_exclude_from_sync
         self.is_mine_id = hs.is_mine_id
         self._clock = hs.get_clock()
+        self._member_info_handler = hs.get_member_info_handler()
 
     async def compute_interested_rooms(
         self,
@@ -645,9 +655,12 @@ class SlidingSyncRoomLists:
                     relevant_room_map[room_id] = room_sync_config
 
         # Filtered subset of `relevant_room_map` for rooms that may have updates
-        # (in the event stream)
-        relevant_rooms_to_send_map = await self._filter_relevant_rooms_to_send(
-            previous_connection_state, from_token, relevant_room_map
+        # (in the event stream, or MSC4428 `member_info` changes)
+        (
+            relevant_rooms_to_send_map,
+            member_info_changes_by_room,
+        ) = await self._filter_relevant_rooms_to_send(
+            previous_connection_state, from_token, to_token, relevant_room_map
         )
 
         return SlidingSyncInterestedRooms(
@@ -659,6 +672,7 @@ class SlidingSyncRoomLists:
             newly_joined_rooms=newly_joined_room_ids,
             newly_left_rooms=set(newly_left_room_map),
             dm_room_ids=dm_room_ids,
+            member_info_changes_by_room=member_info_changes_by_room,
         )
 
     async def _compute_interested_rooms_fallback(
@@ -833,9 +847,12 @@ class SlidingSyncRoomLists:
                     relevant_room_map[room_id] = room_sync_config
 
         # Filtered subset of `relevant_room_map` for rooms that may have updates
-        # (in the event stream)
-        relevant_rooms_to_send_map = await self._filter_relevant_rooms_to_send(
-            previous_connection_state, from_token, relevant_room_map
+        # (in the event stream, or MSC4428 `member_info` changes)
+        (
+            relevant_rooms_to_send_map,
+            member_info_changes_by_room,
+        ) = await self._filter_relevant_rooms_to_send(
+            previous_connection_state, from_token, to_token, relevant_room_map
         )
 
         return SlidingSyncInterestedRooms(
@@ -847,20 +864,29 @@ class SlidingSyncRoomLists:
             newly_joined_rooms=newly_joined_room_ids,
             newly_left_rooms=newly_left_room_ids,
             dm_room_ids=dm_room_ids,
+            member_info_changes_by_room=member_info_changes_by_room,
         )
 
     async def _filter_relevant_rooms_to_send(
         self,
         previous_connection_state: PerConnectionState,
         from_token: StreamToken | None,
+        to_token: StreamToken,
         relevant_room_map: dict[str, RoomSyncConfig],
-    ) -> dict[str, RoomSyncConfig]:
+    ) -> tuple[dict[str, RoomSyncConfig], Mapping[str, Collection[str]] | None]:
         """Filters the `relevant_room_map` down to those rooms that may have
-        updates we need to fetch and return."""
+        updates we need to fetch and return.
+
+        Returns:
+            The filtered map, and the MSC4428 `member_info` changes per room
+            that were looked up to produce it (see
+            `SlidingSyncInterestedRooms.member_info_changes_by_room`).
+        """
 
         # Filtered subset of `relevant_room_map` for rooms that may have updates
         # (in the event stream)
         relevant_rooms_to_send_map: dict[str, RoomSyncConfig] = relevant_room_map
+        member_info_changes_by_room: Mapping[str, Collection[str]] | None = {}
         if relevant_room_map:
             with start_active_span("filter_relevant_rooms_to_send"):
                 if from_token:
@@ -945,13 +971,28 @@ class SlidingSyncRoomLists:
                             raise SlidingSyncUnknownPosition()
 
                     rooms_should_send.update(rooms_that_have_updates)
+
+                    # MSC4428: `member_info` changes with no event anchor are
+                    # not in the event stream, so check their own stream too.
+                    member_info_changes_by_room = (
+                        await self._member_info_handler.get_changes_in_rooms(
+                            relevant_room_map.keys(), from_token, to_token
+                        )
+                    )
+                    if member_info_changes_by_room is None:
+                        # The token predates the retained changes: every room
+                        # may need the full map.
+                        rooms_should_send.update(relevant_room_map.keys())
+                    else:
+                        rooms_should_send.update(member_info_changes_by_room.keys())
+
                     relevant_rooms_to_send_map = {
                         room_id: room_sync_config
                         for room_id, room_sync_config in relevant_room_map.items()
                         if room_id in rooms_should_send
                     }
 
-        return relevant_rooms_to_send_map
+        return relevant_rooms_to_send_map, member_info_changes_by_room
 
     @trace
     async def _get_rewind_changes_to_current_membership_to_token(

synapse/handlers/sync.py

@@ -2827,6 +2827,22 @@ class SyncHandler:
                 sync_result_builder, sync_result_builder.now_token, since_token
             )
 
+        # MSC4428: `member_info` changes with no event anchor since the last
+        # sync, per joined room. `member_info_full` means the since token is
+        # older than the retained changes, so every room gets the full map.
+        member_info_changes_by_room: Mapping[str, Collection[str]] = {}
+        member_info_full = False
+        if since_token and self._member_info_handler.enabled:
+            changes = await self._member_info_handler.get_changes_in_rooms(
+                sync_result_builder.joined_room_ids,
+                since_token,
+                sync_result_builder.now_token,
+            )
+            if changes is None:
+                member_info_full = True
+            else:
+                member_info_changes_by_room = changes
+
         # 2. We check up front if anything has changed, if it hasn't then there is
         # no point in going further.
         #
@@ -2840,6 +2856,7 @@ class SyncHandler:
         # - are there any per-room EDUs;
         # - is there any Room Account Data; or
         # - are there any sticky events in the rooms; or
+        # - are there any `member_info` changes in the rooms; or
         # - might the rooms have changed
         #   (using in-memory event stream change caches, which can
         #   only answer either 'Not changed' or 'Possibly changed')
@@ -2854,6 +2871,8 @@ class SyncHandler:
                 and not ephemeral_by_room
                 and not account_data_by_room
                 and not sticky_by_room
+                and not member_info_changes_by_room
+                and not member_info_full
             ):
                 # This is also a cheap check, but we log the answer
                 have_changed = self._may_have_rooms_changed(sync_result_builder)
@@ -2911,6 +2930,10 @@ class SyncHandler:
                 tags=tags_by_room.get(room_entry.room_id),
                 account_data=account_data_by_room.get(room_entry.room_id, {}),
                 sticky_event_ids=sticky_by_room.get(room_entry.room_id, []),
+                member_info_changes=member_info_changes_by_room.get(
+                    room_entry.room_id, ()
+                ),
+                member_info_full=member_info_full,
                 always_include=sync_result_builder.full_state,
             )
             logger.debug("Generated room entry for %s", room_entry.room_id)
@@ -3296,6 +3319,8 @@ class SyncHandler:
         tags: Mapping[str, JsonMapping] | None,
         account_data: Mapping[str, JsonMapping],
         sticky_event_ids: list[str],
+        member_info_changes: Collection[str] = (),
+        member_info_full: bool = False,
         always_include: bool = False,
     ) -> None:
         """Populates the `joined` and `archived` section of `sync_result_builder`
@@ -3327,6 +3352,10 @@ class SyncHandler:
             account_data: List of new account data for room
             sticky_event_ids: MSC4354 sticky events in the room, if any.
                 In sticky event stream order.
+            member_info_changes: MSC4428: stable IDs whose `member_info` entry
+                changed in this room with no event to anchor the change.
+            member_info_full: MSC4428: send the full `member_info` map for the
+                room (the since token predates the retained changes).
             always_include: Always include this room in the sync response,
                 even if empty.
         """
@@ -3343,6 +3372,8 @@ class SyncHandler:
             or ephemeral
             or full_state
             or sticky_event_ids
+            or member_info_changes
+            or member_info_full
         ):
             if events == [] and tags is None:
                 return
@@ -3436,6 +3467,8 @@ class SyncHandler:
                 or ephemeral
                 or full_state
                 or sticky_event_ids
+                or member_info_changes
+                or member_info_full
             ):
                 return
 
@@ -3518,6 +3551,8 @@ class SyncHandler:
                         (ev.event for ev in sticky_events),
                     ),
                     heroes=summary.get("m.heroes", ()) if summary else (),
+                    changed=member_info_changes,
+                    full=member_info_full,
                 )
 
                 room_sync = JoinedSyncResult(

synapse/notifier.py

@@ -533,7 +533,7 @@ class Notifier:
     @overload
     def on_new_event(
         self,
-        stream_key: Literal[StreamKeyType.RECEIPT],
+        stream_key: Literal[StreamKeyType.RECEIPT, StreamKeyType.MEMBER_INFO],
         new_token: MultiWriterStreamToken,
         users: Collection[str | UserID] | None = None,
         rooms: StrCollection | None = None,

synapse/replication/tcp/client.py

@@ -44,6 +44,7 @@ from synapse.replication.tcp.streams import (
     UnPartialStatedRoomStream,
 )
 from synapse.replication.tcp.streams._base import (
+    MemberInfoChangesStream,
     ProfileUpdatesStream,
     StickyEventsStream,
     ThreadSubscriptionsStream,
@@ -300,6 +301,12 @@ class ReplicationDataHandler:
                 token,
                 rooms=[row.room_id for row in rows],
             )
+        elif stream_name == MemberInfoChangesStream.NAME:
+            self.notifier.on_new_event(
+                StreamKeyType.MEMBER_INFO,
+                self.store.get_member_info_changes_stream_token(),
+                rooms=[row.room_id for row in rows],
+            )
 
         await self._presence_handler.process_replication_rows(
             stream_name, instance_name, token, rows

synapse/replication/tcp/streams/__init__.py

@@ -35,6 +35,7 @@ from synapse.replication.tcp.streams._base import (
     BackfillStream,
     CachesStream,
     DeviceListsStream,
+    MemberInfoChangesStream,
     PresenceFederationStream,
     PresenceStream,
     ProfileUpdatesStream,
@@ -73,6 +74,7 @@ STREAMS_MAP = {
         AccountDataStream,
         ProfileUpdatesStream,
         StickyEventsStream,
+        MemberInfoChangesStream,
         ThreadSubscriptionsStream,
         UnPartialStatedRoomStream,
         UnPartialStatedEventStream,
@@ -98,6 +100,7 @@ __all__ = [
     "AccountDataStream",
     "ProfileUpdatesStream",
     "StickyEventsStream",
+    "MemberInfoChangesStream",
     "ThreadSubscriptionsStream",
     "UnPartialStatedRoomStream",
     "UnPartialStatedEventStream",

synapse/replication/tcp/streams/_base.py

@@ -911,6 +911,48 @@ class StickyEventsStream(_StreamFromIdGen):
         return rows, rows[-1][0], len(updates) == limit
 
 
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class MemberInfoChangesStreamRow:
+    """MSC4428: a room member's `member_info` entry changed without an event
+    in the room to anchor the change (e.g. the user was erased)."""
+
+    room_id: str
+    stable_id: str
+
+
+class MemberInfoChangesStream(_StreamFromIdGen):
+    NAME = "member_info_changes"
+    ROW_TYPE = MemberInfoChangesStreamRow
+
+    def __init__(self, hs: "HomeServer"):
+        self.store = hs.get_datastores().main
+        super().__init__(
+            hs.get_instance_name(),
+            self._update_function,
+            self.store._member_info_changes_id_gen,
+        )
+
+    async def _update_function(
+        self, instance_name: str, from_token: int, to_token: int, limit: int
+    ) -> StreamUpdateResult:
+        updates = await self.store.get_updated_member_info_changes(
+            from_id=from_token, to_id=to_token, limit=limit
+        )
+        rows = [
+            (
+                update.stream_id,
+                # These are the args to `MemberInfoChangesStreamRow`
+                (update.room_id, update.stable_id),
+            )
+            for update in updates
+        ]
+
+        if not rows:
+            return [], to_token, False
+
+        return rows, rows[-1][0], len(updates) == limit
+
+
 @attr.s(slots=True, auto_attribs=True)
 class QuarantinedMediaStreamRow:
     """Row for QuarantinedMediaStream"""

synapse/storage/databases/main/__init__.py

@@ -32,6 +32,7 @@ from synapse.storage.database import (
     LoggingDatabaseConnection,
     LoggingTransaction,
 )
+from synapse.storage.databases.main.member_info import MemberInfoWorkerStore
 from synapse.storage.databases.main.sliding_sync import SlidingSyncStore
 from synapse.storage.databases.main.stats import UserSortOrder
 from synapse.storage.databases.main.sticky_events import StickyEventsWorkerStore
@@ -146,6 +147,7 @@ class DataStore(
     AccountDataStore,
     ThreadSubscriptionsWorkerStore,
     StickyEventsWorkerStore,
+    MemberInfoWorkerStore,
     PushRulesWorkerStore,
     StreamWorkerStore,
     OpenIdStore,

synapse/storage/databases/main/member_info.py

@@ -0,0 +1,289 @@
+#
+# 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>.
+#
+"""MSC4428: the stream of `member_info` changes with no event anchor.
+
+See `synapse.handlers.member_info` for what the map is. Most changes to it are
+delivered alongside the event that caused them; the ones that are not (erasure
+today, link verification once MSC4430 lands) are recorded here, on a stream of
+their own, so that a syncing client is woken and told.
+
+Any worker may write to the stream: it is modelled on the cache invalidation
+stream rather than on a stream with configured writers.
+"""
+
+import logging
+from typing import TYPE_CHECKING, Collection, Mapping, cast
+
+import attr
+
+from synapse.metrics.background_process_metrics import wrap_as_background_process
+from synapse.replication.tcp.streams._base import MemberInfoChangesStream
+from synapse.storage.database import (
+    DatabasePool,
+    LoggingDatabaseConnection,
+    LoggingTransaction,
+)
+from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore
+from synapse.storage.util.id_generators import (
+    MultiWriterIdGenerator,
+    make_multiwriter_sharded_token_bounds_sql,
+)
+from synapse.types import MultiWriterStreamToken
+from synapse.util.duration import Duration
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+logger = logging.getLogger(__name__)
+
+MEMBER_INFO_CHANGES_RETENTION = Duration(days=30)
+"""How long a change row is kept. A client syncing with a token older than this
+is sent the full `member_info` map instead of the deltas."""
+
+MEMBER_INFO_CHANGES_PRUNE_INTERVAL = Duration(hours=1)
+
+
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class MemberInfoChange:
+    stream_id: int
+    room_id: str
+    stable_id: str
+
+
+class MemberInfoWorkerStore(CacheInvalidationWorkerStore):
+    def __init__(
+        self,
+        database: DatabasePool,
+        db_conn: LoggingDatabaseConnection,
+        hs: "HomeServer",
+    ):
+        super().__init__(database, db_conn, hs)
+
+        self._member_info_changes_id_gen: MultiWriterIdGenerator = (
+            MultiWriterIdGenerator(
+                db_conn=db_conn,
+                db=database,
+                notifier=hs.get_replication_notifier(),
+                stream_name="member_info_changes",
+                server_name=self.server_name,
+                instance_name=self._instance_name,
+                tables=[
+                    ("msc4428_member_info_changes", "instance_name", "stream_id"),
+                ],
+                sequence_name="msc4428_member_info_changes_sequence",
+                # Any instance may write to this stream; positions are learnt
+                # from the table (see the module docstring).
+                writers=[],
+            )
+        )
+
+        if (
+            hs.config.worker.run_background_tasks
+            and hs.config.experimental.msc4428_enabled
+        ):
+            self.hs.get_clock().looping_call(
+                self._prune_old_member_info_changes,
+                MEMBER_INFO_CHANGES_PRUNE_INTERVAL,
+            )
+
+    def process_replication_position(
+        self, stream_name: str, instance_name: str, token: int
+    ) -> None:
+        if stream_name == MemberInfoChangesStream.NAME:
+            self._member_info_changes_id_gen.advance(instance_name, token)
+        super().process_replication_position(stream_name, instance_name, token)
+
+    def get_member_info_changes_stream_token(self) -> MultiWriterStreamToken:
+        """The current position of the member info changes stream."""
+        return MultiWriterStreamToken.from_generator(self._member_info_changes_id_gen)
+
+    def get_member_info_changes_stream_id_generator(self) -> MultiWriterIdGenerator:
+        return self._member_info_changes_id_gen
+
+    async def get_rooms_with_membership_for_local_user(self, user_id: str) -> list[str]:
+        """Every room the local user has any membership state in, including
+        rooms they have left, been banned from, or forgotten."""
+        return await self.db_pool.simple_select_onecol(
+            table="local_current_membership",
+            keyvalues={"user_id": user_id},
+            retcol="room_id",
+            desc="get_rooms_with_membership_for_local_user",
+        )
+
+    async def add_member_info_changes(
+        self, changes: Mapping[str, Collection[str]]
+    ) -> MultiWriterStreamToken:
+        """Record that the `member_info` entries of the given stable IDs changed.
+
+        Args:
+            changes: room ID -> the stable IDs whose entry changed in that room.
+
+        Returns:
+            The stream position after the change, to wake syncing clients with.
+        """
+        rows = [
+            (room_id, stable_id)
+            for room_id, stable_ids in changes.items()
+            for stable_id in stable_ids
+        ]
+        if not rows:
+            return self.get_member_info_changes_stream_token()
+
+        now = self.hs.get_clock().time_msec()
+        async with self._member_info_changes_id_gen.get_next_mult(
+            len(rows)
+        ) as stream_ids:
+            await self.db_pool.simple_insert_many(
+                table="msc4428_member_info_changes",
+                keys=(
+                    "stream_id",
+                    "instance_name",
+                    "room_id",
+                    "stable_id",
+                    "changed_ts",
+                ),
+                values=[
+                    (stream_id, self._instance_name, room_id, stable_id, now)
+                    for stream_id, (room_id, stable_id) in zip(stream_ids, rows)
+                ],
+                desc="add_member_info_changes",
+            )
+
+        return self.get_member_info_changes_stream_token()
+
+    async def get_member_info_changes_in_rooms(
+        self,
+        room_ids: Collection[str],
+        from_token: MultiWriterStreamToken,
+        to_token: MultiWriterStreamToken,
+    ) -> dict[str, set[str]] | None:
+        """The stable IDs whose `member_info` entry changed in the given rooms,
+        with stream positions after `from_token` and at or before `to_token`.
+
+        Returns:
+            room ID -> changed stable IDs (only rooms with changes), or None if
+            `from_token` is older than the retained rows, in which case the
+            caller cannot be given deltas and must send the full map.
+        """
+        if not room_ids:
+            return {}
+
+        def _get_member_info_changes_in_rooms_txn(
+            txn: LoggingTransaction,
+        ) -> dict[str, set[str]] | None:
+            max_pruned_stream_id = self._get_max_pruned_member_info_changes_txn(txn)
+            if max_pruned_stream_id > from_token.stream:
+                return None
+
+            # The client may hand us a token ahead of our own position if this
+            # worker is lagging.
+            if to_token.is_before_or_eq(from_token):
+                return {}
+
+            stream_clause, stream_values = make_multiwriter_sharded_token_bounds_sql(
+                self.database_engine,
+                stream_id_column="stream_id",
+                instance_name_column="instance_name",
+                from_token_exclusive=from_token,
+                to_token_inclusive=to_token,
+            )
+            # Rows in the range are few (one per member per anchorless change),
+            # so filter by room here rather than with a long `IN` list.
+            txn.execute(
+                f"""
+                SELECT room_id, stable_id FROM msc4428_member_info_changes
+                WHERE {stream_clause}
+                """,
+                stream_values,
+            )
+
+            changes: dict[str, set[str]] = {}
+            wanted = (
+                room_ids if isinstance(room_ids, (set, frozenset)) else set(room_ids)
+            )
+            for room_id, stable_id in txn:
+                if room_id in wanted:
+                    changes.setdefault(room_id, set()).add(stable_id)
+            return changes
+
+        return await self.db_pool.runInteraction(
+            "get_member_info_changes_in_rooms", _get_member_info_changes_in_rooms_txn
+        )
+
+    def _get_max_pruned_member_info_changes_txn(self, txn: LoggingTransaction) -> int:
+        """The highest stream ID pruned from the change table. Ranges starting
+        at or before it cannot be answered completely."""
+        return self.db_pool.simple_select_one_onecol_txn(
+            txn,
+            table="msc4428_member_info_changes_max_pruned_stream_id",
+            keyvalues={},
+            retcol="stream_id",
+        )
+
+    async def get_updated_member_info_changes(
+        self, from_id: int, to_id: int, limit: int
+    ) -> list[MemberInfoChange]:
+        """Rows with `from_id < stream_id <= to_id`, for replication. Prefer
+        `get_member_info_changes_in_rooms`, which understands sharded tokens."""
+        rows = cast(
+            list[tuple[int, str, str]],
+            await self.db_pool.execute(
+                "get_updated_member_info_changes",
+                """
+                SELECT stream_id, room_id, stable_id FROM msc4428_member_info_changes
+                WHERE ? < stream_id AND stream_id <= ?
+                ORDER BY stream_id ASC
+                LIMIT ?
+                """,
+                from_id,
+                to_id,
+                limit,
+            ),
+        )
+        return [MemberInfoChange(*row) for row in rows]
+
+    @wrap_as_background_process("prune_old_member_info_changes")
+    async def _prune_old_member_info_changes(self) -> None:
+        await self.prune_member_info_changes_before(
+            self.hs.get_clock().time_msec() - MEMBER_INFO_CHANGES_RETENTION.as_millis()
+        )
+
+    async def prune_member_info_changes_before(self, cutoff_ts: int) -> None:
+        """Delete change rows recorded before `cutoff_ts` and record the prune
+        point, so that syncs from older tokens fall back to the full map."""
+
+        def _prune_txn(txn: LoggingTransaction) -> None:
+            txn.execute(
+                "SELECT MAX(stream_id) FROM msc4428_member_info_changes WHERE changed_ts < ?",
+                (cutoff_ts,),
+            )
+            row = txn.fetchone()
+            if row is None or row[0] is None:
+                return
+            max_pruned_stream_id: int = row[0]
+
+            txn.execute(
+                "DELETE FROM msc4428_member_info_changes WHERE changed_ts < ?",
+                (cutoff_ts,),
+            )
+            # Positions only ever advance, whichever worker's clock was ahead.
+            txn.execute(
+                """
+                UPDATE msc4428_member_info_changes_max_pruned_stream_id
+                SET stream_id = ? WHERE stream_id < ?
+                """,
+                (max_pruned_stream_id, max_pruned_stream_id),
+            )
+
+        await self.db_pool.runInteraction("prune_member_info_changes", _prune_txn)

synapse/storage/schema/__init__.py

@@ -177,6 +177,7 @@ Changes in SCHEMA_VERSION = 94
     - MSC4242: Add state DAG tables.
     - MSC4429/MSC4262: Track updates to user profile fields via a new stream.
     - Add an `inserted_ts` column to the `state_groups_persisting` table.
+    - MSC4428: Add the `msc4428_member_info_changes` stream table.
 """
 
 

synapse/storage/schema/main/delta/94/11_msc4428_member_info_changes.sql

@@ -0,0 +1,51 @@
+--
+-- 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>.
+
+-- MSC4428: changes to a room member's `member_info` entry that no event in the
+-- room anchors (today: erasure). Each row says "the entry for `stable_id` in
+-- `room_id` changed", positioned on its own multi-writer stream so that a
+-- syncing client is woken and told about it.
+--
+-- Rows are only kept for a retention period (see `MemberInfoWorkerStore`); the
+-- `_max_pruned_stream_id` table below records how far they have been pruned.
+CREATE TABLE msc4428_member_info_changes (
+  -- Position in the member info changes stream.
+  stream_id BIGINT NOT NULL PRIMARY KEY,
+
+  -- Name of the worker that wrote this row (any worker may write to this
+  -- stream, as for cache invalidations).
+  instance_name TEXT NOT NULL,
+
+  room_id TEXT NOT NULL,
+
+  -- The stable ID (today: user ID) whose entry changed.
+  stable_id TEXT NOT NULL,
+
+  -- When the change was recorded, in milliseconds since the Unix epoch. Used
+  -- to prune old rows.
+  changed_ts BIGINT NOT NULL
+);
+
+CREATE INDEX msc4428_member_info_changes_ts ON msc4428_member_info_changes (changed_ts);
+
+-- The maximum stream_id that has been pruned from `msc4428_member_info_changes`.
+-- A sync token at or before this position cannot be answered from the table,
+-- and the server falls back to sending the full `member_info` map. A separate
+-- table rather than MIN(stream_id), as the device list changes table does, so
+-- that an empty table is not mistaken for "nothing pruned".
+CREATE TABLE msc4428_member_info_changes_max_pruned_stream_id (
+  Lock CHAR(1) NOT NULL DEFAULT 'X' UNIQUE,
+  stream_id BIGINT NOT NULL
+);
+
+INSERT INTO msc4428_member_info_changes_max_pruned_stream_id (stream_id) VALUES (0);

synapse/storage/schema/main/delta/94/11_msc4428_member_info_changes_seq.sql.postgres

@@ -0,0 +1,18 @@
+--
+-- 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>.
+
+CREATE SEQUENCE msc4428_member_info_changes_sequence;
+-- Synapse streams start at 2, because the default position is 1
+-- so any item inserted at position 1 is ignored.
+-- We have to use nextval not START WITH 2, see https://github.com/element-hq/synapse/issues/18712
+SELECT nextval('msc4428_member_info_changes_sequence');

synapse/streams/events.py

@@ -87,6 +87,7 @@ class EventSources:
         sticky_events_key = self.store.get_max_sticky_events_stream_id()
         quarantined_media_key = self.store.get_quarantined_media_stream_token()
         profile_updates_key = self.store.get_max_profile_updates_stream_id()
+        member_info_key = self.store.get_member_info_changes_stream_token()
 
         token = StreamToken(
             room_key=self.sources.room.get_current_key(),
@@ -104,6 +105,7 @@ class EventSources:
             sticky_events_key=sticky_events_key,
             quarantined_media_key=quarantined_media_key,
             profile_updates_key=profile_updates_key,
+            member_info_key=member_info_key,
         )
         return token
 
@@ -134,6 +136,7 @@ class EventSources:
             StreamKeyType.STICKY_EVENTS: self.store.get_sticky_events_stream_id_generator(),
             StreamKeyType.QUARANTINED_MEDIA: self.store.get_quarantined_media_stream_id_generator(),
             StreamKeyType.PROFILE_UPDATES: self.store.get_profile_updates_stream_id_generator(),
+            StreamKeyType.MEMBER_INFO: self.store.get_member_info_changes_stream_id_generator(),
         }
 
         for _, key in StreamKeyType.__members__.items():

synapse/types/__init__.py

@@ -1139,6 +1139,7 @@ class StreamKeyType(Enum):
     STICKY_EVENTS = "sticky_events_key"
     QUARANTINED_MEDIA = "quarantined_media_key"
     PROFILE_UPDATES = "profile_updates_key"
+    MEMBER_INFO = "member_info_key"
 
 
 @attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -1163,6 +1164,7 @@ class StreamToken:
         12. `sticky_events_key`: 4141
         13. `quarantined_media_key`: 4343
         14. `profile_updates_key`: 4444
+        15. `member_info_key`: 4545 (MSC4428; a `MultiWriterStreamToken`)
 
     You can see how many of these keys correspond to the various
     fields in a "/sync" response:
@@ -1227,6 +1229,9 @@ class StreamToken:
         validator=attr.validators.instance_of(MultiWriterStreamToken)
     )
     profile_updates_key: int
+    member_info_key: MultiWriterStreamToken = attr.ib(
+        validator=attr.validators.instance_of(MultiWriterStreamToken)
+    )
 
     _SEPARATOR = "_"
     START: ClassVar["StreamToken"]
@@ -1258,6 +1263,7 @@ class StreamToken:
                 sticky_events_key,
                 quarantined_media_key,
                 profile_updates_key,
+                member_info_key,
             ) = keys
 
             return cls(
@@ -1279,6 +1285,9 @@ class StreamToken:
                     store, quarantined_media_key
                 ),
                 profile_updates_key=int(profile_updates_key),
+                member_info_key=await MultiWriterStreamToken.parse(
+                    store, member_info_key
+                ),
             )
         except CancelledError:
             raise
@@ -1305,6 +1314,7 @@ class StreamToken:
                 str(self.sticky_events_key),
                 await self.quarantined_media_key.to_string(store),
                 str(self.profile_updates_key),
+                await self.member_info_key.to_string(store),
             ]
         )
 
@@ -1340,6 +1350,12 @@ class StreamToken:
                 self.quarantined_media_key.copy_and_advance(new_value),
             )
             return new_token
+        elif key == StreamKeyType.MEMBER_INFO:
+            new_token = self.copy_and_replace(
+                StreamKeyType.MEMBER_INFO,
+                self.member_info_key.copy_and_advance(new_value),
+            )
+            return new_token
 
         new_token = self.copy_and_replace(key, new_value)
         new_id = new_token.get_field(key)
@@ -1363,6 +1379,7 @@ class StreamToken:
             StreamKeyType.RECEIPT,
             StreamKeyType.DEVICE_LIST,
             StreamKeyType.QUARANTINED_MEDIA,
+            StreamKeyType.MEMBER_INFO,
         ],
     ) -> MultiWriterStreamToken: ...
 
@@ -1456,7 +1473,8 @@ class StreamToken:
             f"groups: {self.groups_key}, un_partial_stated_rooms: {self.un_partial_stated_rooms_key}, "
             f"thread_subscriptions: {self.thread_subscriptions_key}, sticky_events: {self.sticky_events_key}, "
             f"quarantined_media: {self.quarantined_media_key}, "
-            f"profile_updates: {self.profile_updates_key})"
+            f"profile_updates: {self.profile_updates_key}, "
+            f"member_info: {self.member_info_key})"
         )
 
 
@@ -1475,6 +1493,7 @@ StreamToken.START = StreamToken(
     sticky_events_key=0,
     quarantined_media_key=MultiWriterStreamToken(stream=0),
     profile_updates_key=0,
+    member_info_key=MultiWriterStreamToken(stream=0),
 )
 
 

tests/replication/tcp/streams/test_member_info.py

@@ -0,0 +1,62 @@
+#
+# 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>.
+#
+
+from synapse.replication.tcp.streams._base import MemberInfoChangesStream
+
+from tests.replication._base import BaseStreamTestCase
+
+
+class MemberInfoChangesStreamTestCase(BaseStreamTestCase):
+    """MSC4428: the member info changes stream reaches workers."""
+
+    def test_changes_are_replicated(self) -> None:
+        store = self.hs.get_datastores().main
+        worker_store = self.worker_hs.get_datastores().main
+
+        self.get_success(
+            store.add_member_info_changes(
+                {
+                    "!a:test": ["@erased:test", "@other:test"],
+                    "!b:test": ["@erased:test"],
+                }
+            )
+        )
+
+        # Not yet connected: no rows should yet have been received
+        self.assertEqual([], self.test_handler.received_rdata_rows)
+
+        self.reconnect()
+        self.replicate()
+
+        received = [
+            (token, row)
+            for stream_name, token, row in self.test_handler.received_rdata_rows
+            if stream_name == MemberInfoChangesStream.NAME
+        ]
+        self.assertEqual(
+            [(row.room_id, row.stable_id) for _, row in received],
+            [
+                ("!a:test", "@erased:test"),
+                ("!a:test", "@other:test"),
+                ("!b:test", "@erased:test"),
+            ],
+        )
+        tokens = [token for token, _ in received]
+        self.assertEqual(tokens, sorted(tokens))
+
+        # The worker's view of the stream position has caught up.
+        self.assertEqual(
+            worker_store.get_member_info_changes_stream_token(),
+            store.get_member_info_changes_stream_token(),
+        )

tests/replication/test_member_info.py

@@ -0,0 +1,81 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+from twisted.internet.testing import MemoryReactor
+
+from synapse.api.constants import MemberInfo
+from synapse.rest import admin
+from synapse.rest.client import login, room, sync
+from synapse.server import HomeServer
+from synapse.types import JsonDict
+from synapse.util.clock import Clock
+
+from tests.replication._base import BaseMultiWorkerStreamTestCase
+from tests.server import make_request
+
+
+class MemberInfoWorkerTestCase(BaseMultiWorkerStreamTestCase):
+    """MSC4428: a `/sync` served by a worker sees an erasure recorded on the
+    main process, with nothing in the room to anchor it."""
+
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+        sync.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        config["experimental_features"] = {"msc4428_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)
+        self.helper.join(self.room_id, self.bob, tok=self.bob_tok)
+
+    def test_erasure_reaches_sync_on_worker(self) -> None:
+        worker_hs = self.make_worker_hs("synapse.app.generic_worker")
+        site = self._hs_to_site[worker_hs]
+
+        channel = make_request(
+            self.reactor, site, "GET", "/sync", access_token=self.alice_tok
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+        self.assertNotIn(
+            MemberInfo.FIELD, channel.json_body["rooms"]["join"][self.room_id]
+        )
+        since = channel.json_body["next_batch"]
+
+        # Erase Bob on the main process, without an event in the room.
+        self.get_success(self.store.mark_user_erased(self.bob))
+        self.get_success(self.hs.get_member_info_handler().on_user_erased(self.bob))
+        self.replicate()
+
+        channel = make_request(
+            self.reactor,
+            site,
+            "GET",
+            f"/sync?since={since}",
+            access_token=self.alice_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+        room = channel.json_body["rooms"]["join"][self.room_id]
+        self.assertEqual(room["timeline"]["events"], [])
+        self.assertEqual(room[MemberInfo.FIELD], {self.bob: {MemberInfo.USER_ID: None}})

tests/rest/admin/test_room.py

@@ -2549,7 +2549,7 @@ class RoomMessagesTestCase(unittest.HomeserverTestCase):
 
     def test_topo_token_is_accepted(self) -> None:
         """Test Topo Token is accepted."""
-        token = "t1-0_0_0_0_0_0_0_0_0_0_0_0_0_0"
+        token = "t1-0_0_0_0_0_0_0_0_0_0_0_0_0_0_0"
         channel = self.make_request(
             "GET",
             "/_synapse/admin/v1/rooms/%s/messages?from=%s" % (self.room_id, token),
@@ -2563,7 +2563,7 @@ class RoomMessagesTestCase(unittest.HomeserverTestCase):
 
     def test_stream_token_is_accepted_for_fwd_pagianation(self) -> None:
         """Test that stream token is accepted for forward pagination."""
-        token = "s0_0_0_0_0_0_0_0_0_0_0_0_0_0"
+        token = "s0_0_0_0_0_0_0_0_0_0_0_0_0_0_0"
         channel = self.make_request(
             "GET",
             "/_synapse/admin/v1/rooms/%s/messages?from=%s" % (self.room_id, token),

tests/rest/client/test_msc4428.py

@@ -304,8 +304,10 @@ class MemberInfoTestCase(unittest.HomeserverTestCase):
         self.helper.send(self.room_id, body="hi", tok=self.bob_tok)
 
     def _erase(self, user_id: str) -> None:
-        """Mark a user erased without deactivating them, so they stay joined."""
+        """Erase a user without deactivating them, so they stay joined: mark
+        them erased and record the change, as deactivation does."""
         self.get_success(self.store.mark_user_erased(user_id))
+        self.get_success(self.hs.get_member_info_handler().on_user_erased(user_id))
 
     def _sync(self, since: str | None = None, **params: str) -> JsonDict:
         query = {**params}
@@ -336,20 +338,74 @@ class MemberInfoTestCase(unittest.HomeserverTestCase):
 
     def test_sync_incremental_only_for_mentioned_members(self) -> None:
         """An incremental sync carries entries only for the stable IDs the
-        response mentions. A change with no event to anchor it (the erasure
-        itself) waits for the next event from, or about, that member."""
+        response mentions, plus any whose entry changed with no event anchor."""
         since = self._sync()["next_batch"]
-        self._erase(self.bob)
 
         # Alice's own message mentions nobody but Alice.
         self.helper.send(self.room_id, body="alice", tok=self.alice_tok)
         body = self._sync(since)
-        room = body["rooms"]["join"][self.room_id]
-        self.assertNotIn(MemberInfo.FIELD, room)
+        self.assertNotIn(MemberInfo.FIELD, body["rooms"]["join"][self.room_id])
         since = body["next_batch"]
 
-        # Bob's next message is the anchor.
+        # Bob's message mentions Bob, but there is nothing to say about him yet.
         self.helper.send(self.room_id, body="bob", tok=self.bob_tok)
+        body = self._sync(since)
+        self.assertNotIn(MemberInfo.FIELD, body["rooms"]["join"][self.room_id])
+        since = body["next_batch"]
+
+        # The erasure has no event anchor: the room comes down with an empty
+        # timeline and the entry.
+        self._erase(self.bob)
+        body = self._sync(since)
+        room = body["rooms"]["join"][self.room_id]
+        self.assertEqual(room["timeline"]["events"], [])
+        self.assertEqual(room[MemberInfo.FIELD], {self.bob: {MemberInfo.USER_ID: None}})
+        since = body["next_batch"]
+
+        # Delivered once: the next sync has nothing to say.
+        self.helper.send(self.room_id, body="alice", tok=self.alice_tok)
+        room = self._sync(since)["rooms"]["join"][self.room_id]
+        self.assertNotIn(MemberInfo.FIELD, room)
+
+    def test_sync_erasure_wakes_long_poll(self) -> None:
+        """A waiting `/sync` returns when a member is erased."""
+        since = self._sync()["next_batch"]
+        channel = self.make_request(
+            "GET",
+            f"/sync?since={since}&timeout=30000",
+            access_token=self.alice_tok,
+            await_result=False,
+        )
+        self.assertFalse(channel.is_finished())
+
+        self._erase(self.bob)
+
+        channel.await_result()
+        self.assertEqual(channel.code, 200, channel.json_body)
+        room = channel.json_body["rooms"]["join"][self.room_id]
+        self.assertEqual(room[MemberInfo.FIELD], {self.bob: {MemberInfo.USER_ID: None}})
+
+    def test_sync_erasure_in_room_member_had_left(self) -> None:
+        """Erasure reaches rooms the user had already left, where no parting
+        leave will ever anchor it (MSC-Q §15: any membership state)."""
+        self.helper.leave(self.room_id, self.bob, tok=self.bob_tok)
+        since = self._sync()["next_batch"]
+
+        self._erase(self.bob)
+
+        room = self._sync(since)["rooms"]["join"][self.room_id]
+        self.assertEqual(room[MemberInfo.FIELD], {self.bob: {MemberInfo.USER_ID: None}})
+
+    def test_sync_token_older_than_retained_changes(self) -> None:
+        """A since token from before the change table was pruned gets the
+        full map rather than (unknowable) deltas."""
+        since = self._sync()["next_batch"]
+        self._erase(self.bob)
+        # Prune everything, as the hourly job would after the retention period.
+        self.get_success(
+            self.store.prune_member_info_changes_before(self.clock.time_msec() + 1)
+        )
+
         room = self._sync(since)["rooms"]["join"][self.room_id]
         self.assertEqual(room[MemberInfo.FIELD], {self.bob: {MemberInfo.USER_ID: None}})
 
@@ -478,13 +534,24 @@ class MemberInfoSlidingSyncTestCase(SlidingSyncBase):
         self.assertTrue(body["rooms"][self.room_id]["initial"])
         self.assertNotIn(MemberInfo.FIELD, body["rooms"][self.room_id])
 
+        # Incrementally: the erasure has no event anchor, but the room comes
+        # down with the entry all the same.
         self.get_success(self.store.mark_user_erased(self.bob))
+        self.get_success(self.hs.get_member_info_handler().on_user_erased(self.bob))
+        body, pos = self.do_sync(self.sync_body, since=pos, tok=self.alice_tok)
+        self.assertNotIn("initial", body["rooms"][self.room_id])
+        self.assertNotIn("timeline", body["rooms"][self.room_id])
+        self.assertEqual(
+            body["rooms"][self.room_id][MemberInfo.FIELD],
+            {self.bob: {MemberInfo.USER_ID: None}},
+        )
 
-        # Incrementally: only once Bob is mentioned again.
+        # Delivered once.
         self.helper.send(self.room_id, body="alice", tok=self.alice_tok)
         body, pos = self.do_sync(self.sync_body, since=pos, tok=self.alice_tok)
         self.assertNotIn(MemberInfo.FIELD, body["rooms"][self.room_id])
 
+        # Mentioning Bob again repeats his entry, as on sync v2.
         self.helper.send(self.room_id, body="bob", tok=self.bob_tok)
         body, pos = self.do_sync(self.sync_body, since=pos, tok=self.alice_tok)
         self.assertEqual(

tests/rest/client/test_rooms.py

@@ -2255,7 +2255,7 @@ class RoomMessageListTestCase(RoomBase):
         self.room_id = self.helper.create_room_as(self.user_id)
 
     def test_topo_token_is_accepted(self) -> None:
-        token = "t1-0_0_0_0_0_0_0_0_0_0_0_0_0_0"
+        token = "t1-0_0_0_0_0_0_0_0_0_0_0_0_0_0_0"
         channel = self.make_request(
             "GET", "/rooms/%s/messages?access_token=x&from=%s" % (self.room_id, token)
         )
@@ -2266,7 +2266,7 @@ class RoomMessageListTestCase(RoomBase):
         self.assertTrue("end" in channel.json_body)
 
     def test_stream_token_is_accepted_for_fwd_pagianation(self) -> None:
-        token = "s0_0_0_0_0_0_0_0_0_0_0_0_0_0"
+        token = "s0_0_0_0_0_0_0_0_0_0_0_0_0_0_0"
         channel = self.make_request(
             "GET", "/rooms/%s/messages?access_token=x&from=%s" % (self.room_id, token)
         )

tests/storage/test_member_info.py

@@ -0,0 +1,99 @@
+#
+# 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>.
+#
+
+from twisted.internet.testing import MemoryReactor
+
+from synapse.server import HomeServer
+from synapse.storage.databases.main.member_info import MEMBER_INFO_CHANGES_RETENTION
+from synapse.util.clock import Clock
+
+from tests import unittest
+
+
+class MemberInfoChangesStoreTestCase(unittest.HomeserverTestCase):
+    """MSC4428: the change table behind the member info changes stream."""
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.store = hs.get_datastores().main
+
+    def test_changes_in_rooms(self) -> None:
+        start = self.store.get_member_info_changes_stream_token()
+
+        # Nothing recorded: nothing to report, for any room.
+        self.assertEqual(
+            self.get_success(
+                self.store.get_member_info_changes_in_rooms({"!a:test"}, start, start)
+            ),
+            {},
+        )
+
+        middle = self.get_success(
+            self.store.add_member_info_changes({"!a:test": ["@u1:test"]})
+        )
+        end = self.get_success(
+            self.store.add_member_info_changes(
+                {"!a:test": ["@u2:test"], "!b:test": ["@u1:test"]}
+            )
+        )
+
+        # Rooms we did not ask about are left out; the range is (from, to].
+        self.assertEqual(
+            self.get_success(
+                self.store.get_member_info_changes_in_rooms({"!a:test"}, start, end)
+            ),
+            {"!a:test": {"@u1:test", "@u2:test"}},
+        )
+        self.assertEqual(
+            self.get_success(
+                self.store.get_member_info_changes_in_rooms(
+                    {"!a:test", "!b:test"}, middle, end
+                )
+            ),
+            {"!a:test": {"@u2:test"}, "!b:test": {"@u1:test"}},
+        )
+        self.assertEqual(
+            self.get_success(
+                self.store.get_member_info_changes_in_rooms({"!b:test"}, end, end)
+            ),
+            {},
+        )
+
+    def test_prune(self) -> None:
+        start = self.store.get_member_info_changes_stream_token()
+        self.get_success(self.store.add_member_info_changes({"!a:test": ["@u1:test"]}))
+        self.reactor.advance(MEMBER_INFO_CHANGES_RETENTION.as_secs() + 1)
+        after_old = self.store.get_member_info_changes_stream_token()
+        end = self.get_success(
+            self.store.add_member_info_changes({"!a:test": ["@u2:test"]})
+        )
+
+        self.get_success(
+            self.store.prune_member_info_changes_before(
+                self.clock.time_msec() - MEMBER_INFO_CHANGES_RETENTION.as_millis()
+            )
+        )
+
+        # A range starting before the prune point cannot be answered.
+        self.assertIsNone(
+            self.get_success(
+                self.store.get_member_info_changes_in_rooms({"!a:test"}, start, end)
+            )
+        )
+        # One starting at it can, and only sees the surviving row.
+        self.assertEqual(
+            self.get_success(
+                self.store.get_member_info_changes_in_rooms({"!a:test"}, after_old, end)
+            ),
+            {"!a:test": {"@u2:test"}},
+        )

tests/test_types.py

@@ -34,6 +34,8 @@ from synapse.types import (
     NonNegativeStrictInt,
     RoomAlias,
     RoomStreamToken,
+    StreamKeyType,
+    StreamToken,
     UserID,
     get_domain_from_id,
     get_localpart_from_id,
@@ -350,3 +352,31 @@ class NonNegativeStrictIntTestCase(unittest.TestCase):
 
         self.assertEqual(MyModel.model_validate_json('{"limit": 0}'), MyModel(limit=0))
         self.assertEqual(MyModel.model_validate({"limit": 42}), MyModel(limit=42))
+
+
+class StreamTokenTestCase(unittest.HomeserverTestCase):
+    """Serialisation of the composite `StreamToken`."""
+
+    def test_round_trip(self) -> None:
+        store = self.hs.get_datastores().main
+
+        token = StreamToken.START.copy_and_replace(
+            StreamKeyType.MEMBER_INFO, MultiWriterStreamToken(stream=7)
+        )
+        string_token = self.get_success(token.to_string(store))
+        # One component per stream, the member info stream last.
+        self.assertEqual(string_token, "s0_0_0_0_0_0_0_0_0_0_0_0_0_0_7")
+
+        parsed = self.get_success(StreamToken.from_string(store, string_token))
+        self.assertEqual(parsed, token)
+
+    def test_parse_without_trailing_components(self) -> None:
+        """Tokens issued before a stream existed parse with that stream at its
+        start position."""
+        store = self.hs.get_datastores().main
+
+        parsed = self.get_success(
+            StreamToken.from_string(store, "s5_0_0_0_0_0_0_0_0_0_0_0_0_0")
+        )
+        self.assertEqual(parsed.room_key, RoomStreamToken(stream=5))
+        self.assertEqual(parsed.member_info_key, MultiWriterStreamToken(stream=0))