From ab8f0607468a2ff7da9f7501084fcbf02cf01255 Mon Sep 17 00:00:00 2001 From: Paul Chobert Date: Tue, 3 Nov 2026 13:14:08 +0100 Subject: [PATCH] Deliver Server Announcement over Legacy Sync (#20412) > [!NOTE] > **Epic**: [Efficient Server Announcements](https://github.com/element-hq/synapse/issues/20240) > Follows: https://github.com/element-hq/synapse/pull/20372 When a new announcement is sent in the announcement room, this PR will ensure that the announcement and the room state are shared with the client over Legacy Sync. If you reviewed the PoC before, please note that there are several differences: - We intentionally don't wake all sync - We don't want typing and receipts - We don't need device lists ### Pull Request Checklist * [x] Pull request is based on the develop branch * [x] Pull request includes a [changelog file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog). The entry should: - Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from `EventStore` to `EventWorkerStore`.". - Use markdown where necessary, mostly for `code blocks`. - End with either a period (.) or an exclamation mark (!). - Start with a capital letter. - Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry. * [x] [Code style](https://element-hq.github.io/synapse/latest/code_style.html) is correct (run the [linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters)) --------- Co-authored-by: Quentin Gliech --- changelog.d/20412.misc | 1 + rust/src/events/mod.rs | 1 + rust/src/events/serialize.rs | 29 +- synapse/api/filtering.py | 3 + synapse/app/generic_worker.py | 4 + synapse/events/utils.py | 17 + synapse/handlers/server_announcements.py | 45 ++ synapse/handlers/sync.py | 219 +++++++- synapse/rest/client/sync.py | 39 +- synapse/storage/databases/main/cache.py | 1 + .../databases/main/server_announcements.py | 44 +- synapse/storage/databases/main/stream.py | 40 ++ synapse/synapse_rust/events.pyi | 18 + tests/handlers/test_server_announcements.py | 38 +- tests/rest/client/test_sync.py | 504 +++++++++++++++++- 15 files changed, 988 insertions(+), 15 deletions(-) create mode 100644 changelog.d/20412.misc diff --git a/changelog.d/20412.misc b/changelog.d/20412.misc new file mode 100644 index 0000000000..00a9b4f284 --- /dev/null +++ b/changelog.d/20412.misc @@ -0,0 +1 @@ +Deliver server announcement rooms to users who are not members of them in legacy `/sync`. diff --git a/rust/src/events/mod.rs b/rust/src/events/mod.rs index 3fc2d5ad69..6e0dc95135 100644 --- a/rust/src/events/mod.rs +++ b/rust/src/events/mod.rs @@ -119,6 +119,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> child_module.add_function(wrap_pyfunction!(redact_event_py, m)?)?; child_module.add_function(wrap_pyfunction!(redact_event_dict, m)?)?; child_module.add_function(wrap_pyfunction!(serialize::serialize_events, m)?)?; + child_module.add_function(wrap_pyfunction!(serialize::serialize_synthetic_event, m)?)?; child_module.add_function(wrap_pyfunction!(serialize::format_event_raw, m)?)?; child_module.add_function(wrap_pyfunction!(serialize::format_event_for_client_v1, m)?)?; child_module.add_function(wrap_pyfunction!(serialize::format_event_for_client_v2, m)?)?; diff --git a/rust/src/events/serialize.rs b/rust/src/events/serialize.rs index 20d4d9bf52..0d1a0fda91 100644 --- a/rust/src/events/serialize.rs +++ b/rust/src/events/serialize.rs @@ -36,7 +36,7 @@ use pyo3::{ types::{PyAnyMethods, PyDict, PyDictMethods}, Bound, PyAny, PyResult, Python, }; -use pythonize::pythonize; +use pythonize::{depythonize, pythonize}; use serde_json::{Map, Number, Value}; use crate::{ @@ -636,6 +636,33 @@ fn format_for_client_v2(d: &mut Map) { } } +/// Serialize an event that only exists as JSON, such as a synthetic event that +/// is never persisted. +/// +/// Only the event format and the field allowlist from `config` apply: there is +/// no stored event to take the age, transaction ID, redaction or bundled +/// aggregations from. +#[pyfunction] +pub fn serialize_synthetic_event<'py>( + py: Python<'py>, + event: Bound<'py, PyAny>, + config: &SerializeEventConfig, +) -> PyResult> { + let mut d: Map = depythonize(&event)?; + + if config.as_client_event { + apply_event_format(config.event_format, &mut d); + } + + if let Some(fields) = &config.event_field_allowlist { + if !fields.is_empty() { + d = only_fields(&d, fields)?; + } + } + + Ok(pythonize(py, &Value::Object(d))?) +} + // Standalone versions of the client format transforms, re-exported from // `synapse.events.utils` purely as a backwards compatibility hack: they have // never been part of the module API and modules shouldn't be pulling them in, diff --git a/synapse/api/filtering.py b/synapse/api/filtering.py index cbae9133c6..0d7b63ef07 100644 --- a/synapse/api/filtering.py +++ b/synapse/api/filtering.py @@ -292,6 +292,9 @@ class FilterCollection: await self._room_filter.filter(events) ) + def filter_rooms(self, room_ids: Iterable[str]) -> set[str]: + return self._room_filter.filter_rooms(room_ids) + def blocks_all_rooms(self) -> bool: return self._room_filter.filters_all_rooms() diff --git a/synapse/app/generic_worker.py b/synapse/app/generic_worker.py index 6014d328ba..bddc3334e3 100644 --- a/synapse/app/generic_worker.py +++ b/synapse/app/generic_worker.py @@ -98,6 +98,9 @@ from synapse.storage.databases.main.relations import RelationsWorkerStore from synapse.storage.databases.main.room import RoomWorkerStore from synapse.storage.databases.main.roommember import RoomMemberWorkerStore from synapse.storage.databases.main.search import SearchStore +from synapse.storage.databases.main.server_announcements import ( + ServerAnnouncementsStore, +) from synapse.storage.databases.main.session import SessionStore from synapse.storage.databases.main.signatures import SignatureWorkerStore from synapse.storage.databases.main.sliding_sync import SlidingSyncStore @@ -171,6 +174,7 @@ class GenericWorkerStore( ExperimentalFeaturesStore, SlidingSyncStore, DelayedEventsStore, + ServerAnnouncementsStore, ): # Properties that multiple storage classes define. Tell mypy what the # expected type is. diff --git a/synapse/events/utils.py b/synapse/events/utils.py index 365b7b1559..d6a7d8cb6c 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -51,6 +51,7 @@ from synapse.synapse_rust.events import ( format_event_raw, redact_event, serialize_events, + serialize_synthetic_event, ) from synapse.synapse_rust.types import Requester from synapse.types import JsonDict @@ -423,6 +424,22 @@ class EventClientSerializer: ) return [event for event in stitched if event is not None] + def serialize_synthetic_event( + self, event: JsonDict, config: SerializeEventConfig + ) -> JsonDict: + """Serializes an event that only exists as JSON, such as a synthetic + event that is never persisted, applying the event format and field + allowlist from `config`. + + Args: + event: The event's JSON, in federation format. + config: Event serialization config + + Returns: + The serialized event + """ + return serialize_synthetic_event(event, config) + def register_add_extra_fields_to_unsigned_client_event_callback( self, callback: ADD_EXTRA_FIELDS_TO_UNSIGNED_CLIENT_EVENT_CALLBACK ) -> None: diff --git a/synapse/handlers/server_announcements.py b/synapse/handlers/server_announcements.py index a8a4b7f78c..6adfe49ee3 100644 --- a/synapse/handlers/server_announcements.py +++ b/synapse/handlers/server_announcements.py @@ -20,12 +20,57 @@ from synapse.config import ConfigError from synapse.config.server_announcements import ServerAnnouncementRoom from synapse.types import JsonDict, UserID, create_requester from synapse.types.state import StateFilter +from synapse.util.hash import sha256_and_url_safe_base64 if TYPE_CHECKING: from synapse.server import HomeServer logger = logging.getLogger(__name__) +SYNTHETIC_JOIN_EVENT_ID_PREFIX = "$fake_announcement_join" + + +def make_synthetic_join_event( + room_id: str, user_id: str, origin_server_ts: int +) -> JsonDict: + """Build the join event shown to a user who is not actually in an announcement + room, so that clients list the room as joined. + + The event is never persisted. Its event ID is derived from the room and user, + so the same user always receives the same event and clients drop repeats. The + prefix makes it obvious that the event ID is not a real one. + + It is injected into the `/sync` responses of every user of the homeserver who + is not one of the room's senders. + + Args: + room_id: The announcement room. + user_id: The user receiving the event. + origin_server_ts: The timestamp to give the event. Pass the same value + every time, e.g. that of the room's create event. + + Returns: + The event's JSON in federation format. + """ + return { + "room_id": room_id, + "type": EventTypes.Member, + "event_id": "-".join( + [ + SYNTHETIC_JOIN_EVENT_ID_PREFIX, + # 8 first chars of room_id and user_id localpart for debugging + room_id[1:9], + UserID.from_string(user_id).localpart[:8], + # hash of room_id + user_id for uniqueness & staying under 255 bytes length + sha256_and_url_safe_base64(f"{room_id}\0{user_id}"), + ] + ), + "sender": user_id, + "state_key": user_id, + "origin_server_ts": origin_server_ts, + "content": {"membership": Membership.JOIN}, + } + class ServerAnnouncementsHandler: """Reconcile configured rooms on the main process, without deleting removed entries.""" diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index c35e8f01fb..b4b260d156 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -276,6 +276,7 @@ class JoinedSyncResult: unread_thread_notifications: JsonDict summary: JsonDict | None unread_count: int + is_announcement: bool = False def __bool__(self) -> bool: """Make the result appear empty if there are no updates. This is used @@ -2364,12 +2365,27 @@ class SyncHandler: if room_id not in mutable_rooms_to_exclude ) + last_membership_by_room_id = { + event.room_id: event.membership for event in membership_change_events + } + synthetic_join_announcement_room_ids = frozenset( + room_id + for room_id in sync_config.filter_collection.filter_rooms( + await self.store.get_server_announcement_room_ids() + ) + if room_id not in mutable_joined_room_ids + and room_id not in mutable_rooms_to_exclude + and last_membership_by_room_id.get(room_id) + in (None, Membership.LEAVE, Membership.BAN) + ) + sync_result_builder = SyncResultBuilder( sync_config, full_state, since_token=since_token, now_token=now_token, joined_room_ids=joined_room_ids, + synthetic_join_announcement_room_ids=synthetic_join_announcement_room_ids, excluded_room_ids=frozenset(mutable_rooms_to_exclude), forced_newly_joined_room_ids=frozenset(forced_newly_joined_room_ids), membership_change_events=membership_change_events, @@ -2966,7 +2982,9 @@ class SyncHandler: ignored_users = await self.store.ignored_users(user_id) if since_token: room_changes = await self._get_room_changes_for_incremental_sync( - sync_result_builder, ignored_users + sync_result_builder, + ignored_users, + account_data_room_ids=account_data_by_room.keys(), ) tags_by_room = await self.store.get_updated_tags( user_id, since_token.account_data_key @@ -3034,7 +3052,10 @@ class SyncHandler: return True stream_id = since_token.room_key.stream - for room_id in sync_result_builder.joined_room_ids: + for room_id in ( + sync_result_builder.joined_room_ids + | sync_result_builder.synthetic_join_announcement_room_ids + ): if self.store.has_room_changed_since(room_id, stream_id): return True return False @@ -3043,6 +3064,7 @@ class SyncHandler: self, sync_result_builder: "SyncResultBuilder", ignored_users: frozenset[str], + account_data_room_ids: AbstractSet[str], ) -> _RoomChanges: """Determine the changes in rooms to report to the user. @@ -3081,6 +3103,7 @@ class SyncHandler: sync_result_builder.forced_newly_joined_room_ids ) newly_left_rooms: list[str] = [] + left_announcement_room_ids: set[str] = set() room_entries: list[RoomSyncResultBuilder] = [] invited: list[InvitedSyncResult] = [] knocked: list[KnockedSyncResult] = [] @@ -3184,6 +3207,10 @@ class SyncHandler: if knock_room_sync: knocked.append(knock_room_sync) + if room_id in sync_result_builder.synthetic_join_announcement_room_ids: + left_announcement_room_ids.add(room_id) + continue + # Always include leave/ban events. Just take the last one. # TODO: How do we handle ban -> leave in same batch? leave_events = [ @@ -3291,6 +3318,15 @@ class SyncHandler: room_entries.append(entry) + room_entries.extend( + await self._get_announcement_room_entries( + sync_result_builder, + sync_result_builder.synthetic_join_announcement_room_ids, + left_room_ids=left_announcement_room_ids, + account_data_room_ids=account_data_room_ids, + ) + ) + return _RoomChanges( room_entries, invited, @@ -3299,6 +3335,144 @@ class SyncHandler: newly_left_rooms, ) + async def _get_announcement_room_entries( + self, + sync_result_builder: "SyncResultBuilder", + room_ids: AbstractSet[str], + left_room_ids: AbstractSet[str] = frozenset(), + account_data_room_ids: AbstractSet[str] = frozenset(), + ) -> list["RoomSyncResultBuilder"]: + """Returns entries for the given announcement rooms that have something + to show. + + A room only shows up once it has had an announcement, i.e. a non-state + event, so that creating the room and joining its senders does not show + it to every user. On an incremental sync, a room is included if it has + had an announcement and there is any new event in it since the + `since_token`, such as a sender leaving or a name change, or a change + to the user's room account data for it. With `full_state` set, a room + is also included without any new event. On an + initial sync, a room is included if it has ever had an announcement, + even when later state changes push that announcement out of the + timeline. + + Rooms in `left_room_ids` are included even without an announcement, so + that the client replaces the user's real membership with the synthetic + join and gets the room's current state. + + An entry carries the full state only if any of: + - there is no `since_token` + - the first announcement is after the `since_token` + - the user left or was removed from the room + Otherwise it carries the changes since the `since_token`, as for a + joined room: any sync covering a room's first announcement delivers it + with full state, so the client already has the room. + + Sending an announcement does not wake up waiting syncs, so that an + announcement does not wake every user's sync at once. A waiting sync + picks announcements up when it next runs: at its timeout, or when + something else wakes it. With no other activity, an announcement is up + to one sync timeout late. It can be up to about two timeouts late when + something woke the sync without producing anything for it: the sync + then returns that empty result at its timeout without running again, + and the announcement waits for the next sync. + + Args: + sync_result_builder + room_ids: The announcement rooms to consider. + left_room_ids: Rooms among `room_ids` that the user left or was + removed from since the `since_token` of `sync_result_builder`. + account_data_room_ids: Rooms whose room account data for the user + changed since the `since_token` of `sync_result_builder`. + They may include rooms that are not in `room_ids`. + """ + since_token = sync_result_builder.since_token + now_token = sync_result_builder.now_token + timeline_limit = ( + sync_result_builder.sync_config.filter_collection.timeline_limit() + ) + + room_to_events = await self.store.get_room_events_stream_for_rooms( + room_ids=room_ids, + from_key=now_token.room_key, + to_key=since_token.room_key if since_token else None, + limit=timeline_limit + 1, + direction=Direction.BACKWARDS, + ) + + # Rooms with an announcement since the `since_token`. + announced_room_ids = { + room_id + for room_id, (events, _, _) in room_to_events.items() + if not all(event.is_state() for event in events) + } + + # Rooms the user left are included even without an announcement, so + # their real membership is replaced by the synthetic join. + room_ids_to_include = announced_room_ids | left_room_ids + + # On an incremental sync, `room_ids_to_include` should also include + # rooms that have new state events but no new announcement in the + # window, or a room account data change, if they had an announcement + # before the window. + # + # On an initial or `full_state` sync, a room is included whenever it + # has ever had an announcement, even if it isn't among the recent + # events fetched above. + if not since_token or sync_result_builder.full_state: + room_ids_to_check = room_ids - room_ids_to_include + else: + room_ids_to_check = { + room_id + for room_id, (events, _, _) in room_to_events.items() + if events and room_id not in left_room_ids + } + # Also consider rooms whose account data has changed. + room_ids_to_check |= account_data_room_ids & ( + room_ids - room_ids_to_include + ) + first_announcement_positions = ( + await self.store.get_first_non_state_event_positions( + room_ids_to_check, now_token.room_key.get_max_stream_pos() + ) + ) + room_ids_to_include.update(first_announcement_positions) + + room_entries = [] + for room_id in room_ids_to_include: + if room_id in room_to_events: + events, start_key, _ = room_to_events[room_id] + events.reverse() + upto_token = now_token.copy_and_replace(StreamKeyType.ROOM, start_key) + else: + # Without a `since_token`, `room_to_events` has an entry for every + # room in `room_ids`. + assert since_token is not None + events = [] + upto_token = since_token + + first_announcement_position = first_announcement_positions.get(room_id) + full_state = ( + since_token is None + or first_announcement_position is None + or first_announcement_position.persisted_after(since_token.room_key) + ) + + room_entries.append( + RoomSyncResultBuilder( + room_id=room_id, + rtype="joined", + events=events, + newly_joined=False, + full_state=full_state, + since_token=since_token, + upto_token=upto_token, + end_token=now_token, + ) + ) + + return room_entries + async def _get_room_changes_for_initial_sync( self, sync_result_builder: "SyncResultBuilder", @@ -3360,6 +3534,14 @@ class SyncHandler: knock = await self.store.get_event(event.event_id) knocked.append(KnockedSyncResult(room_id=event.room_id, knock=knock)) elif event.membership in (Membership.LEAVE, Membership.BAN): + # Former members of an announcement room get the synthetic join + # instead of this leave or ban. + if ( + event.room_id + in sync_result_builder.synthetic_join_announcement_room_ids + ): + continue + # Always send down rooms we were banned from or kicked from. if not sync_config.filter_collection.include_leave: if event.membership == Membership.LEAVE: @@ -3382,6 +3564,19 @@ class SyncHandler: ) ) + room_ids_in_response = ( + {entry.room_id for entry in room_entries} + | {result.room_id for result in invited} + | {result.room_id for result in knocked} + ) + room_entries.extend( + await self._get_announcement_room_entries( + sync_result_builder, + sync_result_builder.synthetic_join_announcement_room_ids + - room_ids_in_response, + ) + ) + return _RoomChanges(room_entries, invited, knocked, [], []) async def _generate_room_entry( @@ -3615,10 +3810,19 @@ class SyncHandler: summary=summary, unread_count=0, sticky=sticky_events, + is_announcement=( + room_id + in sync_result_builder.synthetic_join_announcement_room_ids + ), ) if room_sync or always_include: - notifs = await self.unread_notifs_for_room_id(room_id, sync_config) + if room_sync.is_announcement: + notifs = RoomNotifCounts.empty() + else: + notifs = await self.unread_notifs_for_room_id( + room_id, sync_config + ) # Notifications for the main timeline. notify_count = notifs.main_timeline.notify_count @@ -3805,6 +4009,11 @@ class SyncResultBuilder: since_token: The token supplied by user, or None. now_token: The token to sync up to. joined_room_ids: List of rooms the user is joined to + synthetic_join_announcement_room_ids: Server announcement rooms the user + is not a member of, whose new announcements should still be sent with + a synthetic join. Rooms the user really joined are not included. + They are kept apart from `joined_room_ids` because most code that + reads `joined_room_ids` relies on the user being a member. excluded_room_ids: Set of room ids we should omit from the /sync response. forced_newly_joined_room_ids: Rooms that should be presented in the /sync response as if they were @@ -3829,6 +4038,7 @@ class SyncResultBuilder: since_token: StreamToken | None now_token: StreamToken joined_room_ids: frozenset[str] + synthetic_join_announcement_room_ids: frozenset[str] excluded_room_ids: frozenset[str] forced_newly_joined_room_ids: frozenset[str] membership_change_events: list[EventBase] @@ -3855,6 +4065,9 @@ class SyncResultBuilder: newly_left_users = set() if self.since_token: for joined_sync in self.joined: + if joined_sync.room_id in self.synthetic_join_announcement_room_ids: + continue + it = itertools.chain( joined_sync.state.values(), (e.event for e in joined_sync.timeline.events), diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py index 6e6eb6d633..da56f46632 100644 --- a/synapse/rest/client/sync.py +++ b/synapse/rest/client/sync.py @@ -24,7 +24,13 @@ from typing import TYPE_CHECKING, Any, Literal, Mapping import attr -from synapse.api.constants import AccountDataTypes, EduTypes, Membership, PresenceState +from synapse.api.constants import ( + AccountDataTypes, + EduTypes, + EventTypes, + Membership, + PresenceState, +) from synapse.api.errors import Codes, StoreError, SynapseError from synapse.api.filtering import FilterCollection from synapse.api.presence import UserPresenceState @@ -37,6 +43,7 @@ from synapse.events.utils import ( strip_event, ) from synapse.handlers.presence import format_user_presence_state +from synapse.handlers.server_announcements import make_synthetic_join_event from synapse.handlers.sliding_sync import SlidingSyncConfig, SlidingSyncResult from synapse.handlers.sync import ( ArchivedSyncResult, @@ -598,6 +605,26 @@ class SyncRestServlet(RestServlet): state_dict = room.state timeline_events = room.timeline.events + user_id = sync_config.user.to_string() + is_announcement = isinstance(room, JoinedSyncResult) and room.is_announcement + # A former member gets the room with `is_announcement` set, as they are + # no longer a member, but their real leave or ban can still be in the + # state or the timeline. Remove it, as the synthetic join replaces it. + if is_announcement: + state_dict = { + key: event + for key, event in state_dict.items() + if key != (EventTypes.Member, user_id) + } + timeline_events = [ + filtered_event + for filtered_event in timeline_events + if not ( + filtered_event.event.type == EventTypes.Member + and filtered_event.event.state_key == user_id + ) + ] + state_events = state_dict.values() for event in state_events: @@ -626,6 +653,16 @@ class SyncRestServlet(RestServlet): time_now, config=serialize_options, ) + if is_announcement: + create_event = await self.store.get_create_event_for_room(room.room_id) + serialized_state.append( + self._event_serializer.serialize_synthetic_event( + make_synthetic_join_event( + room.room_id, user_id, create_event.origin_server_ts + ), + serialize_options, + ) + ) serialized_timeline = await self._event_serializer.serialize_events( timeline_events, time_now, diff --git a/synapse/storage/databases/main/cache.py b/synapse/storage/databases/main/cache.py index 45fe605d8b..5ca908ac5b 100644 --- a/synapse/storage/databases/main/cache.py +++ b/synapse/storage/databases/main/cache.py @@ -612,6 +612,7 @@ class CacheInvalidationWorkerStore(SQLBaseStore): self._attempt_to_invalidate_cache("get_room_version_id", (room_id,)) self._attempt_to_invalidate_cache("get_room_type", (room_id,)) self._attempt_to_invalidate_cache("get_room_encryption", (room_id,)) + self._attempt_to_invalidate_cache("get_server_announcement_room_ids", None) self._attempt_to_invalidate_cache("_get_max_event_pos", (room_id,)) diff --git a/synapse/storage/databases/main/server_announcements.py b/synapse/storage/databases/main/server_announcements.py index bb466ca97a..444fd9a12f 100644 --- a/synapse/storage/databases/main/server_announcements.py +++ b/synapse/storage/databases/main/server_announcements.py @@ -12,17 +12,18 @@ # . # -from synapse.storage._base import SQLBaseStore +from synapse.storage.database import LoggingTransaction +from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore +from synapse.util.caches.descriptors import cached -class ServerAnnouncementsStore(SQLBaseStore): +class ServerAnnouncementsStore(CacheInvalidationWorkerStore): """Stores details about an announcement room. The unique announcement `id` identifies the same room even when its configured details change. It is separate from the actual room ID. - Lookups are cheap and deliberately uncached, avoiding cross-worker cache - invalidation for mapping and avatar changes. + Only the set of configured room IDs is cached, as it is read on every sync. """ async def get_server_announcement_room(self, config_id: str) -> str | None: @@ -45,14 +46,41 @@ class ServerAnnouncementsStore(SQLBaseStore): desc="get_server_announcement_id", ) + @cached(num_args=0) + async def get_server_announcement_room_ids(self) -> frozenset[str]: + """Return the room IDs of the currently configured announcement rooms. + + Rooms whose config ID is no longer in the config are left out, as + are configured rooms that have not been created yet. + """ + rows = await self.db_pool.simple_select_many_batch( + table="server_announcement_rooms", + column="config_id", + iterable=[ + room.config_id for room in self.hs.config.serverannouncements.rooms + ], + retcols=("room_id",), + desc="get_server_announcement_room_ids", + ) + return frozenset(room_id for (room_id,) in rows) + async def store_server_announcement_room( self, config_id: str, room_id: str ) -> None: """Store a new mapping.""" - await self.db_pool.simple_insert( - table="server_announcement_rooms", - values={"config_id": config_id, "room_id": room_id}, - desc="store_server_announcement_room", + + def _store_server_announcement_room_txn(txn: LoggingTransaction) -> None: + self.db_pool.simple_insert_txn( + txn, + table="server_announcement_rooms", + values={"config_id": config_id, "room_id": room_id}, + ) + self._invalidate_cache_and_stream( + txn, self.get_server_announcement_room_ids, () + ) + + await self.db_pool.runInteraction( + "store_server_announcement_room", _store_server_announcement_room_txn ) async def get_server_announcement_avatar( diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 59048c2fc6..d278985832 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1841,6 +1841,46 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore): return results + async def get_first_non_state_event_positions( + self, room_ids: Collection[str], max_stream_ordering: int + ) -> dict[str, PersistedEventPosition]: + """Return the position of the first non-state, non-outlier event at or + before `max_stream_ordering` in each of `room_ids`, leaving out rooms that + have none. + """ + + def _get_first_non_state_event_positions_txn( + txn: LoggingTransaction, + ) -> dict[str, PersistedEventPosition]: + positions = {} + for room_id in room_ids: + txn.execute( + """ + SELECT instance_name, stream_ordering FROM events + WHERE room_id = ? AND stream_ordering <= ? + AND state_key IS NULL + AND NOT outlier + ORDER BY stream_ordering + LIMIT 1 + """, + (room_id, max_stream_ordering), + ) + row = txn.fetchone() + if row is not None: + instance_name, stream_ordering = row + positions[room_id] = PersistedEventPosition( + instance_name or "master", stream_ordering + ) + return positions + + if not room_ids: + return {} + + return await self.db_pool.runInteraction( + "get_first_non_state_event_positions", + _get_first_non_state_event_positions_txn, + ) + async def get_current_room_stream_token_for_room_id( self, room_id: str ) -> RoomStreamToken: diff --git a/synapse/synapse_rust/events.pyi b/synapse/synapse_rust/events.pyi index bac7af5a13..1690d71753 100644 --- a/synapse/synapse_rust/events.pyi +++ b/synapse/synapse_rust/events.pyi @@ -482,6 +482,24 @@ def serialize_events( The serialized events, in the same order as `events`. """ +def serialize_synthetic_event( + event: JsonDict, config: SerializeEventConfig +) -> JsonDict: + """Serialize an event that only exists as JSON, such as a synthetic event + that is never persisted. + + Only the event format and the field allowlist from `config` apply: there is + no stored event to take the age, transaction ID, redaction or bundled + aggregations from. + + Args: + event: The event's JSON, in federation format. + config: The serialization config. + + Returns: + The serialized event. + """ + # The standalone `format_event_*` transforms below are a backwards # compatibility hack: they have never been part of the module API and modules # shouldn't be pulling them in, but some in the wild import them (via diff --git a/tests/handlers/test_server_announcements.py b/tests/handlers/test_server_announcements.py index c3d423746e..d69fcecf1c 100644 --- a/tests/handlers/test_server_announcements.py +++ b/tests/handlers/test_server_announcements.py @@ -21,7 +21,10 @@ from synapse.api.constants import EventTypes from synapse.api.errors import SynapseError from synapse.config import ConfigError from synapse.config.server_announcements import ServerAnnouncementRoom -from synapse.handlers.server_announcements import ServerAnnouncementsHandler +from synapse.handlers.server_announcements import ( + ServerAnnouncementsHandler, + make_synthetic_join_event, +) from synapse.rest import admin from synapse.rest.client import login, register, room, sync from synapse.server import HomeServer @@ -350,3 +353,36 @@ class ServerAnnouncementsTestCase(unittest.HomeserverTestCase): self.get_success(self.store.get_server_announcement_room("updates")), room_id, ) + + +class SyntheticJoinEventTestCase(unittest.TestCase): + def test_event_id(self) -> None: + """The event ID is the prefix, the first 8 characters of the room ID and + of the user localpart, and a hash of both IDs. + """ + self.assertEqual( + make_synthetic_join_event( + "!Q8bOeGjhyXNtCVmX7Zvvh-W5KiVu2Q8vBK_2ljoZW7s", + "@alexander:example.com", + 0, + )["event_id"], + "$fake_announcement_join-Q8bOeGjh-alexande-SVJqkL5sx8x0TXrBVujJXBUw_Kc-c10PNz_sZV5FTnk", + ) + + def test_event_id_short_localpart(self) -> None: + """A localpart shorter than 8 characters is included in full.""" + self.assertEqual( + make_synthetic_join_event( + "!Q8bOeGjhyXNtCVmX7Zvvh-W5KiVu2Q8vBK_2ljoZW7s", "@bob:example.com", 0 + )["event_id"], + "$fake_announcement_join-Q8bOeGjh-bob-OkzlBBrE1j2T2pq5aF4ubO-Z1hmIo2cun6rrIXjw2QE", + ) + + def test_event_id_room_id_with_domain(self) -> None: + """For a room ID with a domain, only the start of its localpart is included.""" + self.assertEqual( + make_synthetic_join_event( + "!abcdefghijklmnopqr:example.com", "@alexander:example.com", 0 + )["event_id"], + "$fake_announcement_join-abcdefgh-alexande-QdoyrQ6WLSZZBDWVTw4cVhH61QX8Z90v8YIhfWkHTlg", + ) diff --git a/tests/rest/client/test_sync.py b/tests/rest/client/test_sync.py index 4f80758033..418c2810dd 100644 --- a/tests/rest/client/test_sync.py +++ b/tests/rest/client/test_sync.py @@ -21,6 +21,7 @@ import json import logging +import attr from parameterized import parameterized from twisted.internet.testing import MemoryReactor @@ -34,8 +35,19 @@ from synapse.api.constants import ( ReceiptTypes, RelationTypes, ) +from synapse.config.server_announcements import ServerAnnouncementRoom from synapse.events import EventBase -from synapse.rest.client import devices, knock, login, read_marker, receipts, room, sync +from synapse.handlers.server_announcements import SYNTHETIC_JOIN_EVENT_ID_PREFIX +from synapse.rest.client import ( + account_data, + devices, + knock, + login, + read_marker, + receipts, + room, + sync, +) from synapse.server import HomeServer from synapse.types import JsonDict, RoomStreamToken, StreamKeyType, StreamToken from synapse.util.clock import Clock @@ -2187,3 +2199,493 @@ class SyncLazyLoadedMembersCacheTestCase(unittest.HomeserverTestCase): if e["type"] == EventTypes.Member ] self.assertIn(self.alice, members) + + +class ServerAnnouncementsSyncTestCase(unittest.HomeserverTestCase): + """Tests that server announcement rooms are delivered to users who are not + members of them, with a synthetic join for the syncing user. + """ + + servlets = [ + synapse.rest.admin.register_servlets, + login.register_servlets, + room.register_servlets, + sync.register_servlets, + account_data.register_servlets, + ] + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.announcer = self.register_user("announcer", "pass") + self.announcer_tok = self.login("announcer", "pass") + self.former = self.register_user("former", "pass") + self.former_tok = self.login("former", "pass") + self.bob = self.register_user("bob", "pass") + self.bob_tok = self.login("bob", "pass") + + self.config = hs.config.serverannouncements + self.config.rooms = [ + ServerAnnouncementRoom( + config_id="news", + room_name="Server News", + room_topic="News from the server", + room_avatar_url=None, + room_avatar_path=None, + allowed_user_ids=[self.announcer, self.former], + ) + ] + self.handler = hs.get_server_announcements_handler() + self.store = hs.get_datastores().main + self.get_success(self.handler.reconcile_rooms()) + room_id = self.get_success(self.store.get_server_announcement_room("news")) + assert room_id is not None + self.room_id = room_id + + def _sync( + self, + tok: str, + since: str | None = None, + full_state: bool = False, + filter: JsonDict | None = None, + use_state_after: bool = False, + ) -> JsonDict: + params = [] + if since is not None: + params.append(f"since={since}") + if full_state: + params.append("full_state=true") + if use_state_after: + params.append("use_state_after=true") + if filter is not None: + params.append(f"filter={json.dumps(filter)}") + channel = self.make_request( + "GET", "/sync?" + "&".join(params), access_token=tok + ) + self.assertEqual(channel.code, 200, channel.json_body) + return channel.json_body + + def _announce(self, body: str) -> str: + return self.helper.send(self.room_id, body, tok=self.announcer_tok)["event_id"] + + def _assert_not_delivered(self, response: JsonDict) -> None: + for section in ("join", "invite", "leave", "knock"): + self.assertNotIn( + self.room_id, response.get("rooms", {}).get(section, {}), section + ) + + def _assert_delivered(self, response: JsonDict, user_id: str) -> JsonDict: + """Assert that the announcement room is in the `join` section with a + synthetic join for `user_id` and no unread counts, and return it. + """ + room = response["rooms"]["join"][self.room_id] + state = room["state_after"] if "state_after" in room else room["state"] + + own_memberships = [ + event + for event in state["events"] + room["timeline"]["events"] + if event["type"] == EventTypes.Member and event["state_key"] == user_id + ] + self.assertEqual(len(own_memberships), 1, own_memberships) + membership = own_memberships[0] + self.assertTrue( + membership["event_id"].startswith(SYNTHETIC_JOIN_EVENT_ID_PREFIX), + membership, + ) + self.assertEqual(membership["content"], {"membership": Membership.JOIN}) + create_event = self.get_success( + self.store.get_create_event_for_room(self.room_id) + ) + self.assertEqual(membership["origin_server_ts"], create_event.origin_server_ts) + + self.assertEqual( + room["unread_notifications"], + {"highlight_count": 0, "notification_count": 0}, + ) + return room + + def _timeline_ids(self, room: JsonDict) -> list[str]: + return [event["event_id"] for event in room["timeline"]["events"]] + + def _timeline_types(self, room: JsonDict) -> list[str]: + return [event["type"] for event in room["timeline"]["events"]] + + def _state_types(self, room: JsonDict) -> set[str]: + return {event["type"] for event in room["state"]["events"]} + + def test_incremental_sync_delivers_new_announcement(self) -> None: + """A non-member syncing incrementally gets an announcement sent since + their token, with the room's full state. + """ + since = self._sync(self.bob_tok)["next_batch"] + event_id = self._announce("hello") + + room = self._assert_delivered(self._sync(self.bob_tok, since), self.bob) + self.assertEqual(self._timeline_ids(room), [event_id]) + self.assertTrue( + {EventTypes.Create, EventTypes.Name, EventTypes.Topic} + <= self._state_types(room) + ) + + def test_incremental_sync_without_new_announcement(self) -> None: + """An announcement room with nothing new since the token is left out of + an incremental sync. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + + self._assert_not_delivered(self._sync(self.bob_tok, since)) + + def test_incremental_sync_after_first_announcement(self) -> None: + """A non-member who has synced past an announcement room's first + announcement gets a later announcement with only the synthetic join in + `state`, as nothing else changed since their token. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + event_id = self._announce("again") + + room = self._assert_delivered(self._sync(self.bob_tok, since), self.bob) + self.assertEqual(self._timeline_ids(room), [event_id]) + self.assertEqual( + [event["type"] for event in room["state"]["events"]], [EventTypes.Member] + ) + + def test_members_get_their_real_membership(self) -> None: + """Members of an announcement room sync it as a regular room.""" + self._announce("hello") + + room = self._sync(self.announcer_tok)["rooms"]["join"][self.room_id] + memberships = [ + event + for event in room["state"]["events"] + room["timeline"]["events"] + if event["type"] == EventTypes.Member + ] + self.assertIn(self.announcer, {event["state_key"] for event in memberships}) + for event in memberships: + self.assertFalse( + event["event_id"].startswith(SYNTHETIC_JOIN_EVENT_ID_PREFIX), event + ) + + def test_full_state_sync_delivers_earlier_announcement(self) -> None: + """An incremental sync with `full_state` delivers the state of an + announcement room even when there is no announcement since the token. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + + room = self._assert_delivered( + self._sync(self.bob_tok, since, full_state=True), self.bob + ) + self.assertEqual(self._timeline_ids(room), []) + self.assertTrue( + {EventTypes.Create, EventTypes.Name, EventTypes.Topic} + <= self._state_types(room) + ) + + def test_full_state_sync_without_announcement(self) -> None: + """An announcement room without any announcement is left out of an + incremental sync with `full_state`. + """ + since = self._sync(self.bob_tok)["next_batch"] + + self._assert_not_delivered(self._sync(self.bob_tok, since, full_state=True)) + + def test_initial_sync_delivers_announcements(self) -> None: + """An initial sync delivers an announcement room with its most recent + announcements. + """ + event_ids = [self._announce(f"message {i}") for i in range(3)] + + room = self._assert_delivered( + self._sync(self.bob_tok, filter={"room": {"timeline": {"limit": 2}}}), + self.bob, + ) + self.assertEqual(self._timeline_ids(room), event_ids[-2:]) + self.assertTrue(room["timeline"]["limited"]) + + def test_state_changes_after_last_announcement(self) -> None: + """An announcement room is delivered on initial, incremental and + `full_state` syncs even when state changes push its last announcement + out of the timeline. The incremental sync has the state changes in its + timeline. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + + self.config.rooms = [ + attr.evolve( + self.config.rooms[0], room_name="New name", room_topic="New topic" + ) + ] + self.get_success(self.handler.reconcile_rooms()) + filter = {"room": {"timeline": {"limit": 1}}} + + for response in ( + self._sync(self.bob_tok, filter=filter), + self._sync(self.bob_tok, since, filter=filter), + self._sync(self.bob_tok, since, full_state=True, filter=filter), + ): + room = self._assert_delivered(response, self.bob) + self.assertNotIn(EventTypes.Message, self._timeline_types(room)) + topic = [ + event + for event in room["state"]["events"] + room["timeline"]["events"] + if event["type"] == EventTypes.Topic + ][-1] + self.assertEqual(topic["content"]["topic"], "New topic") + + room = self._assert_delivered(self._sync(self.bob_tok, since), self.bob) + self.assertIn(EventTypes.Topic, self._timeline_types(room)) + + def test_state_change_after_announcement(self) -> None: + """Once an announcement room has had an announcement, a state change + such as a sender leaving delivers the room to non-members with the + change in the timeline, as it does for members. + """ + self._announce("hello") + bob_since = self._sync(self.bob_tok)["next_batch"] + announcer_since = self._sync(self.announcer_tok)["next_batch"] + + leave_id = self.helper.leave(self.room_id, self.former, tok=self.former_tok)[ + "event_id" + ] + + room = self._assert_delivered(self._sync(self.bob_tok, bob_since), self.bob) + self.assertEqual(self._timeline_ids(room), [leave_id]) + room = self._sync(self.announcer_tok, announcer_since)["rooms"]["join"][ + self.room_id + ] + self.assertEqual(self._timeline_ids(room), [leave_id]) + + def test_state_change_before_first_announcement(self) -> None: + """Before an announcement room's first announcement, a state change such + as a sender leaving does not deliver the room on incremental sync. + """ + since = self._sync(self.bob_tok)["next_batch"] + + self.helper.leave(self.room_id, self.former, tok=self.former_tok) + + self._assert_not_delivered(self._sync(self.bob_tok, since)) + + def test_initial_sync_without_announcement(self) -> None: + """An announcement room without any announcement is left out of an + initial sync. + """ + self._assert_not_delivered(self._sync(self.bob_tok)) + + def test_filtered_out_room(self) -> None: + """An announcement room excluded by the sync filter is not delivered.""" + self._announce("hello") + filter = {"room": {"not_rooms": [self.room_id]}} + + since = self._sync(self.bob_tok, filter=filter)["next_batch"] + self._announce("again") + + self._assert_not_delivered(self._sync(self.bob_tok, filter=filter)) + self._assert_not_delivered(self._sync(self.bob_tok, since, filter=filter)) + self._assert_not_delivered( + self._sync(self.bob_tok, since, full_state=True, filter=filter) + ) + + def test_removed_member(self) -> None: + """A member removed from the room by the config switches straight to the + synthetic join and the room's current power levels, then gets later + announcements as a non-member, on initial sync too. + """ + since = self._sync(self.former_tok)["next_batch"] + + self.config.rooms = [ + attr.evolve(self.config.rooms[0], allowed_user_ids=[self.announcer]) + ] + self.get_success(self.handler.reconcile_rooms()) + response = self._sync(self.former_tok, since) + room = self._assert_delivered(response, self.former) + self.assertNotIn(EventTypes.Message, self._timeline_types(room)) + power_levels = [ + event + for event in room["state"]["events"] + room["timeline"]["events"] + if event["type"] == EventTypes.PowerLevels + ][-1] + self.assertNotIn(self.former, power_levels["content"]["users"]) + + event_id = self._announce("hello") + room = self._assert_delivered( + self._sync(self.former_tok, response["next_batch"]), self.former + ) + self.assertEqual(self._timeline_ids(room), [event_id]) + + room = self._assert_delivered(self._sync(self.former_tok), self.former) + self.assertEqual(self._timeline_ids(room)[-1], event_id) + + def test_removed_member_syncing_after_announcements(self) -> None: + """A member removed from the room by the config while offline gets the + announcements sent since their removal on their next incremental sync. + """ + since = self._sync(self.former_tok)["next_batch"] + + self.config.rooms = [ + attr.evolve(self.config.rooms[0], allowed_user_ids=[self.announcer]) + ] + self.get_success(self.handler.reconcile_rooms()) + event_ids = [self._announce(f"message {i}") for i in range(2)] + + response = self._sync(self.former_tok, since) + self.assertNotIn(self.room_id, response["rooms"].get("leave", {})) + room = self._assert_delivered(response, self.former) + self.assertEqual(self._timeline_ids(room)[-2:], event_ids) + + def test_member_leaving(self) -> None: + """A member who leaves the room on their own switches straight to the + synthetic join, even when no announcement is sent. + """ + since = self._sync(self.former_tok)["next_batch"] + + self.helper.leave(self.room_id, self.former, tok=self.former_tok) + + room = self._assert_delivered(self._sync(self.former_tok, since), self.former) + self.assertEqual(self._timeline_ids(room), []) + + def test_initial_sync_after_leaving(self) -> None: + """A former member who left on their own gets a single, synthetic join + on initial sync, even when their real leave is among the recent events. + """ + self.helper.leave(self.room_id, self.former, tok=self.former_tok) + self._announce("hello") + + self._assert_delivered(self._sync(self.former_tok), self.former) + + def test_no_device_list_changes(self) -> None: + """Delivering an announcement room does not report its members as users + whose device lists changed. + """ + since = self._sync(self.bob_tok)["next_batch"] + self._announce("hello") + + response = self._sync(self.bob_tok, since) + self._assert_delivered(response, self.bob) + self.assertEqual(response.get("device_lists", {}).get("changed", []), []) + + def test_marked_unread_without_new_events(self) -> None: + """A non-member who marks an announcement room as unread gets + `m.marked_unread` on their next incremental sync, even when nothing + else happened in the room. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + + channel = self.make_request( + "PUT", + f"/user/{self.bob}/rooms/{self.room_id}/account_data/m.marked_unread", + {"unread": True}, + access_token=self.bob_tok, + ) + self.assertEqual(channel.code, 200, channel.json_body) + + room = self._assert_delivered(self._sync(self.bob_tok, since), self.bob) + self.assertEqual(self._timeline_ids(room), []) + self.assertEqual( + room["account_data"]["events"], + [{"type": "m.marked_unread", "content": {"unread": True}}], + ) + + def test_marked_unread_with_new_announcement(self) -> None: + """A non-member who marks an announcement room as unread, followed by a + new announcement, gets both the announcement and `m.marked_unread` in + the room on their next incremental sync. + """ + self._announce("hello") + since = self._sync(self.bob_tok)["next_batch"] + + channel = self.make_request( + "PUT", + f"/user/{self.bob}/rooms/{self.room_id}/account_data/m.marked_unread", + {"unread": True}, + access_token=self.bob_tok, + ) + self.assertEqual(channel.code, 200, channel.json_body) + event_id = self._announce("again") + + room = self._assert_delivered(self._sync(self.bob_tok, since), self.bob) + self.assertEqual(self._timeline_ids(room), [event_id]) + self.assertEqual( + room["account_data"]["events"], + [{"type": "m.marked_unread", "content": {"unread": True}}], + ) + + def test_state_after(self) -> None: + """With `use_state_after`, a non-member gets the synthetic join in + `state_after`, including a former member whose real leave is in the + window. + """ + bob_since = self._sync(self.bob_tok, use_state_after=True)["next_batch"] + former_since = self._sync(self.former_tok, use_state_after=True)["next_batch"] + + self.helper.leave(self.room_id, self.former, tok=self.former_tok) + self._announce("hello") + + for tok, user_id, since in ( + (self.bob_tok, self.bob, bob_since), + (self.former_tok, self.former, former_since), + ): + room = self._assert_delivered( + self._sync(tok, since, use_state_after=True), user_id + ) + self.assertNotIn("state", room) + + def test_lazy_loaded_members(self) -> None: + """The synthetic join is sent when members are lazy-loaded.""" + self._announce("hello") + + self._assert_delivered( + self._sync( + self.bob_tok, filter={"room": {"state": {"lazy_load_members": True}}} + ), + self.bob, + ) + + def test_event_fields(self) -> None: + """With an `event_fields` filter, the synthetic join is trimmed to the + requested fields like every other event. + """ + self._announce("hello") + event_fields = ["type", "content", "state_key"] + + response = self._sync(self.bob_tok, filter={"event_fields": event_fields}) + room = response["rooms"]["join"][self.room_id] + + events = room["state"]["events"] + room["timeline"]["events"] + own_memberships = [ + event + for event in events + if event["type"] == EventTypes.Member and event["state_key"] == self.bob + ] + self.assertEqual( + own_memberships, + [ + { + "type": EventTypes.Member, + "content": {"membership": Membership.JOIN}, + "state_key": self.bob, + } + ], + ) + for event in events: + self.assertLessEqual(set(event), set(event_fields), event) + + def test_federation_event_format(self) -> None: + """With the `federation` event format, the synthetic join carries the + room ID like every other event. + """ + self._announce("hello") + + response = self._sync(self.bob_tok, filter={"event_format": "federation"}) + room = response["rooms"]["join"][self.room_id] + + events = room["state"]["events"] + room["timeline"]["events"] + own_memberships = [ + event + for event in events + if event["type"] == EventTypes.Member and event["state_key"] == self.bob + ] + self.assertEqual(len(own_memberships), 1, own_memberships) + for event in events: + self.assertEqual(event.get("room_id"), self.room_id, event)