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

<!-- Please read
https://element-hq.github.io/synapse/latest/development/contributing_guide.html
before submitting your pull request -->

* [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 <quenting@element.io>
This commit is contained in:
Paul ChobertandQuentin Gliech authored and GitHub committed 2026-11-03 13:14:08 +01:00
1 parent 241dc05caa
commit ab8f060746
15 files changed
+988 -15

No files matched your search

+1
View File
@@ -0,0 +1 @@
Deliver server announcement rooms to users who are not members of them in legacy `/sync`.
+1
View File
@@ -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)?)?;
+28 -1
View File
@@ -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<String, Value>) {
}
}
/// 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<Bound<'py, PyAny>> {
let mut d: Map<String, Value> = 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,
+3
View File
@@ -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()
+4
View File
@@ -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.
+17
View File
@@ -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:
+45
View File
@@ -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."""
+216 -3
View File
@@ -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),
+38 -1
View File
@@ -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,
+1
View File
@@ -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,))
@@ -12,17 +12,18 @@
# <https://www.gnu.org/licenses/agpl-3.0.html>.
#
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(
+40
View File
@@ -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:
+18
View File
@@ -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
+37 -1
View File
@@ -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",
)
+503 -1
View File
@@ -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)