diff --git a/synapse/_scripts/synapse_port_db.py b/synapse/_scripts/synapse_port_db.py index f4f598d27c..c37b235a30 100755 --- a/synapse/_scripts/synapse_port_db.py +++ b/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", diff --git a/synapse/app/generic_worker.py b/synapse/app/generic_worker.py index 6014d328ba..82a4542f49 100644 --- a/synapse/app/generic_worker.py +++ b/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, diff --git a/synapse/handlers/deactivate_account.py b/synapse/handlers/deactivate_account.py index a1aa2a51dc..ba5b77313c 100644 --- a/synapse/handlers/deactivate_account.py +++ b/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. diff --git a/synapse/handlers/member_info.py b/synapse/handlers/member_info.py index 91443cecd8..1400ff7023 100644 --- a/synapse/handlers/member_info.py +++ b/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 diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 99c326e062..3aa0d91eb2 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/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( diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index f4d1f484e7..7f3ee4877a 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/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( diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 5be0f39f93..94c65192bf 100644 --- a/synapse/handlers/sync.py +++ b/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( diff --git a/synapse/notifier.py b/synapse/notifier.py index 3ee0fc60d6..54c46d4fcd 100644 --- a/synapse/notifier.py +++ b/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, diff --git a/synapse/replication/tcp/client.py b/synapse/replication/tcp/client.py index 8a60f8df6f..be0a7dd0c7 100644 --- a/synapse/replication/tcp/client.py +++ b/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 diff --git a/synapse/replication/tcp/streams/__init__.py b/synapse/replication/tcp/streams/__init__.py index e657822da7..340123014c 100644 --- a/synapse/replication/tcp/streams/__init__.py +++ b/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", diff --git a/synapse/replication/tcp/streams/_base.py b/synapse/replication/tcp/streams/_base.py index f9c7821a39..0c8c6bab58 100644 --- a/synapse/replication/tcp/streams/_base.py +++ b/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""" diff --git a/synapse/storage/databases/main/__init__.py b/synapse/storage/databases/main/__init__.py index 9f8d4debbe..51751232b6 100644 --- a/synapse/storage/databases/main/__init__.py +++ b/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, diff --git a/synapse/storage/databases/main/member_info.py b/synapse/storage/databases/main/member_info.py new file mode 100644 index 0000000000..082606178d --- /dev/null +++ b/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: +# . +# +"""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) diff --git a/synapse/storage/schema/__init__.py b/synapse/storage/schema/__init__.py index 926121d8e1..15236197ac 100644 --- a/synapse/storage/schema/__init__.py +++ b/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. """ diff --git a/synapse/storage/schema/main/delta/94/11_msc4428_member_info_changes.sql b/synapse/storage/schema/main/delta/94/11_msc4428_member_info_changes.sql new file mode 100644 index 0000000000..bb0ae8174f --- /dev/null +++ b/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: +-- . + +-- 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); diff --git a/synapse/storage/schema/main/delta/94/11_msc4428_member_info_changes_seq.sql.postgres b/synapse/storage/schema/main/delta/94/11_msc4428_member_info_changes_seq.sql.postgres new file mode 100644 index 0000000000..190b218136 --- /dev/null +++ b/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: +-- . + +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'); diff --git a/synapse/streams/events.py b/synapse/streams/events.py index 24120eb736..40b046cce2 100644 --- a/synapse/streams/events.py +++ b/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(): diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index 90d6c9f32f..ae0bb0b541 100644 --- a/synapse/types/__init__.py +++ b/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), ) diff --git a/tests/replication/tcp/streams/test_member_info.py b/tests/replication/tcp/streams/test_member_info.py new file mode 100644 index 0000000000..eb3855098c --- /dev/null +++ b/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: +# . +# + +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(), + ) diff --git a/tests/replication/test_member_info.py b/tests/replication/test_member_info.py new file mode 100644 index 0000000000..9b98d68961 --- /dev/null +++ b/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: +# . +# + +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}}) diff --git a/tests/rest/admin/test_room.py b/tests/rest/admin/test_room.py index b776929e21..c6cf84c975 100644 --- a/tests/rest/admin/test_room.py +++ b/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), diff --git a/tests/rest/client/test_msc4428.py b/tests/rest/client/test_msc4428.py index 28a03eb494..9ca20704be 100644 --- a/tests/rest/client/test_msc4428.py +++ b/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( diff --git a/tests/rest/client/test_rooms.py b/tests/rest/client/test_rooms.py index c5a01a9a7b..2c386f0052 100644 --- a/tests/rest/client/test_rooms.py +++ b/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) ) diff --git a/tests/storage/test_member_info.py b/tests/storage/test_member_info.py new file mode 100644 index 0000000000..a231c772b1 --- /dev/null +++ b/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: +# . +# + +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"}}, + ) diff --git a/tests/test_types.py b/tests/test_types.py index 43fd96d6f5..b621b2b36e 100644 --- a/tests/test_types.py +++ b/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))