diff --git a/Cargo.lock b/Cargo.lock index 5faf827e1d..5f2835ce4c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -564,9 +564,9 @@ dependencies = [ [[package]] name = "http" -version = "1.4.0" +version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a" +checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" dependencies = [ "bytes", "itoa", @@ -946,9 +946,9 @@ dependencies = [ [[package]] name = "log" -version = "0.4.29" +version = "0.4.32" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +checksum = "953f07c43838f8e6f9758cab68bf5bed85465e7587ebe0b823f1bcd81978ad3a" [[package]] name = "lru-slab" @@ -1692,9 +1692,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.149" +version = "1.0.150" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" dependencies = [ "itoa", "memchr", diff --git a/changelog.d/19591.feature b/changelog.d/19591.feature new file mode 100644 index 0000000000..f780030807 --- /dev/null +++ b/changelog.d/19591.feature @@ -0,0 +1 @@ +Expose [MSC4354 Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) over [MSC4186 (Simplified) Sliding Sync](https://github.com/matrix-org/matrix-spec-proposals/pull/4186). \ No newline at end of file diff --git a/changelog.d/19785.bugfix b/changelog.d/19785.bugfix new file mode 100644 index 0000000000..643481ee27 --- /dev/null +++ b/changelog.d/19785.bugfix @@ -0,0 +1 @@ +Fix a long-standing bug where the badge notification count for a room could become permanently inflated if a read receipt was sent before the room's notification counts were first summarised. diff --git a/changelog.d/19829.doc b/changelog.d/19829.doc new file mode 100644 index 0000000000..44dbc0eff4 --- /dev/null +++ b/changelog.d/19829.doc @@ -0,0 +1 @@ +Tweak wording of Rust crate dependency update policy. \ No newline at end of file diff --git a/docs/deprecation_policy.md b/docs/deprecation_policy.md index 2a4301b4ee..3ce64f01d7 100644 --- a/docs/deprecation_policy.md +++ b/docs/deprecation_policy.md @@ -79,11 +79,17 @@ concerned about the criteria for selecting minimum versions. The only thing of c is making sure we're not making it unnecessarily difficult for downstream package maintainers. Generally, this just means avoiding the bleeding edge for a few months. -The situation for Rust dependencies is fundamentally different. For packagers, the -concerns around Python dependency versions do not apply. The `cargo` tool handles -downloading and building all libraries to satisfy dependencies, and these libraries are -statically linked into the final binary. This means that from a packager's perspective, -the Rust dependency versions are an internal build detail, not a runtime dependency to -be managed on the target system. Consequently, we have even greater flexibility to -upgrade Rust dependencies as needed for the project. Some distros (e.g. Fedora) do -package Rust libraries, but this appears to be the outlier rather than the norm. +The situation for Rust dependencies is typically different and +the concerns around Python dependency versions typically do not apply. +For example, for packagers of Debian, the packagers have the choice of either +using a crate packaged in the distro, or vendoring crates in the source package for the +application. + +This freedom to vendor a dependency crate for a specific application consequently gives +us even greater flexibility to upgrade Rust dependencies as needed for the project. + +(This is in contrast with Python dependencies, which are generally +installed system-wide by mainstream distributions' official packages.) + +Some distros (e.g. Fedora) do not vendor Rust dependencies in their +official application packages, but these cases appear to be less common. diff --git a/synapse/handlers/delayed_events.py b/synapse/handlers/delayed_events.py index 6afb7c0f3c..f016d95e31 100644 --- a/synapse/handlers/delayed_events.py +++ b/synapse/handlers/delayed_events.py @@ -36,6 +36,7 @@ from synapse.storage.databases.main.delayed_events import ( ) from synapse.storage.databases.main.state_deltas import StateDelta from synapse.types import ( + Absent, JsonDict, Requester, RoomID, @@ -45,7 +46,6 @@ from synapse.types import ( from synapse.util.duration import Duration from synapse.util.events import generate_fake_event_id from synapse.util.metrics import Measure -from synapse.util.sentinel import Sentinel if TYPE_CHECKING: from synapse.server import HomeServer @@ -273,9 +273,7 @@ class DelayedEventsHandler: ) continue - sender_str = event_id_and_sender_dict.get( - delta.event_id, Sentinel.UNSET_SENTINEL - ) + sender_str = event_id_and_sender_dict.get(delta.event_id, Absent) if sender_str is None: # An event exists, but the `sender` field was "null" and Synapse # incorrectly accepted the event. This is not expected. @@ -285,7 +283,7 @@ class DelayedEventsHandler: delta.event_id, ) continue - if sender_str is Sentinel.UNSET_SENTINEL: + if sender_str is Absent: # We have an event ID, but the event was not found in the # datastore. This can happen if a room, or its history, is # purged. State deltas related to the room are left behind, but diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index e7f0e7340b..10ca3ddea0 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -287,7 +287,6 @@ class SlidingSyncHandler: lists = interested_rooms.lists relevant_room_map = interested_rooms.relevant_room_map - all_rooms = interested_rooms.all_rooms room_membership_for_user_map = interested_rooms.room_membership_for_user_map relevant_rooms_to_send_map = interested_rooms.relevant_rooms_to_send_map @@ -328,6 +327,7 @@ class SlidingSyncHandler: actual_lists=lists, previous_connection_state=previous_connection_state, new_connection_state=new_connection_state, + all_interested_room_ids=interested_rooms.all_rooms, # We're purposely using `relevant_room_map` instead of # `relevant_rooms_to_send_map` here. This needs to be all room_ids we could # send regardless of whether they have an event update or not. The @@ -350,7 +350,7 @@ class SlidingSyncHandler: if from_token: # The set of rooms that the client (may) care about, but aren't # in any list range (or subscribed to). - missing_rooms = all_rooms - relevant_room_map.keys() + missing_rooms = interested_rooms.all_rooms - relevant_room_map.keys() # We now just go and try fetching any events in the above rooms # to see if anything has happened since the `from_token`. diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index 4a324e9661..b3342de778 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/synapse/handlers/sliding_sync/extensions.py @@ -11,7 +11,6 @@ # See the GNU Affero General Public License for more details: # . # - import itertools import logging from collections import ChainMap @@ -26,11 +25,13 @@ from typing import ( from typing_extensions import TypeAlias, assert_never -from synapse.api.constants import AccountDataTypes, EduTypes +from synapse.api.constants import AccountDataTypes, EduTypes, StickyEvent +from synapse.events.utils import FilteredEvent from synapse.handlers.receipts import ReceiptEventSource from synapse.logging.opentracing import trace from synapse.storage.databases.main.receipts import ReceiptInRoom from synapse.types import ( + Absent, DeviceListUpdates, JsonMapping, MultiWriterStreamToken, @@ -47,10 +48,12 @@ from synapse.types.handlers.sliding_sync import ( SlidingSyncConfig, SlidingSyncResult, ) +from synapse.types.rest.client import SlidingSyncStickyEventsToken from synapse.util.async_helpers import ( concurrently_execute, gather_optional_coroutines, ) +from synapse.visibility import filter_and_transform_events_for_client _ThreadSubscription: TypeAlias = ( SlidingSyncResult.Extensions.ThreadSubscriptionsExtension.ThreadSubscription @@ -73,7 +76,10 @@ class SlidingSyncExtensionHandler: self.event_sources = hs.get_event_sources() self.device_handler = hs.get_device_handler() self.push_rules_handler = hs.get_push_rules_handler() + self.clock = hs.get_clock() + self._storage_controllers = hs.get_storage_controllers() self._enable_thread_subscriptions = hs.config.experimental.msc4306_enabled + self._enable_sticky_events = hs.config.experimental.msc4354_enabled @trace async def get_extensions_response( @@ -81,6 +87,7 @@ class SlidingSyncExtensionHandler: sync_config: SlidingSyncConfig, previous_connection_state: "PerConnectionState", new_connection_state: "MutablePerConnectionState", + all_interested_room_ids: set[str], actual_lists: Mapping[str, SlidingSyncResult.SlidingWindowList], actual_room_ids: set[str], actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult], @@ -91,9 +98,12 @@ class SlidingSyncExtensionHandler: Args: sync_config: Sync configuration - new_connection_state: Snapshot of the current per-connection state - new_per_connection_state: A mutable copy of the per-connection + previous_connection_state: Snapshot of the current per-connection state + new_connection_state: A mutable copy of the per-connection state, used to record updates to the state during this request. + all_interested_room_ids: The IDs of all rooms that the client is interested in, + even if they don't appear in the current limited window. + See `SlidingSyncInterestedRooms.all_rooms`. actual_lists: Sliding window API. A map of list key to list results in the Sliding Sync response. actual_room_ids: The actual room IDs in the the Sliding Sync response. @@ -174,6 +184,19 @@ class SlidingSyncExtensionHandler: from_token=from_token, ) + sticky_events_coro = None + if ( + sync_config.extensions.sticky_events is not Absent + and self._enable_sticky_events + ): + sticky_events_coro = self.get_sticky_events_extension_response( + sync_config=sync_config, + sticky_events_request=sync_config.extensions.sticky_events, + all_interested_room_ids=all_interested_room_ids, + to_token=to_token, + from_token=from_token, + ) + ( to_device_response, e2ee_response, @@ -181,6 +204,7 @@ class SlidingSyncExtensionHandler: receipts_response, typing_response, thread_subs_response, + sticky_events_response, ) = await gather_optional_coroutines( to_device_coro, e2ee_coro, @@ -188,6 +212,7 @@ class SlidingSyncExtensionHandler: receipts_coro, typing_coro, thread_subs_coro, + sticky_events_coro, ) return SlidingSyncResult.Extensions( @@ -197,6 +222,7 @@ class SlidingSyncExtensionHandler: receipts=receipts_response, typing=typing_response, thread_subscriptions=thread_subs_response, + sticky_events=sticky_events_response, ) def find_relevant_room_ids_for_extension( @@ -967,3 +993,65 @@ class SlidingSyncExtensionHandler: unsubscribed=unsubscribed_threads, prev_batch=prev_batch, ) + + async def get_sticky_events_extension_response( + self, + sync_config: SlidingSyncConfig, + sticky_events_request: SlidingSyncConfig.Extensions.StickyEventsExtension, + all_interested_room_ids: set[str], + to_token: StreamToken, + from_token: SlidingSyncStreamToken | None, + ) -> SlidingSyncResult.Extensions.StickyEventsExtension | None: + if not sticky_events_request.enabled: + return None + now = self.clock.time_msec() + # If there is no `since` token specified, start from the beginning of the stream + # to make sure the client receives all visible (unexpired) sticky events + since_token = sticky_events_request.since or SlidingSyncStickyEventsToken.START + ( + sticky_events_to_id, + room_to_event_ids, + ) = await self.store.get_sticky_events_in_rooms( + all_interested_room_ids, + from_id=since_token.sticky_events_stream_id, + to_id=to_token.sticky_events_key, + now=now, + limit=min(sticky_events_request.limit, StickyEvent.MAX_EVENTS_IN_SYNC), + ) + # No need to preserve sticky event order here because we will + # reassemble it in the right order after. + all_sticky_event_ids = { + ev_id for evs in room_to_event_ids.values() for ev_id in evs + } + unfiltered_events = await self.store.get_events_as_list(all_sticky_event_ids) + filtered_events = await filter_and_transform_events_for_client( + self._storage_controllers, + sync_config.user.to_string(), + unfiltered_events, + # As per MSC4354: + # > History visibility checks MUST NOT be applied to sticky events. + # > Any joined user is authorised to see sticky events for the duration they remain sticky. + always_include_ids=frozenset(all_sticky_event_ids), + ) + filtered_event_map = {ev.event.event_id: ev for ev in filtered_events} + + room_id_to_sticky_events: dict[str, list[FilteredEvent]] = {} + for room_id, sticky_event_ids in room_to_event_ids.items(): + filtered_events_for_room = [ + filtered_event_map[event_id] + # This reintroduces the correct order + # (by the sticky events stream) + for event_id in sticky_event_ids + if event_id in filtered_event_map + ] + if len(filtered_events_for_room) == 0: + continue + + room_id_to_sticky_events[room_id] = filtered_events_for_room + + return SlidingSyncResult.Extensions.StickyEventsExtension( + room_id_to_sticky_events=room_id_to_sticky_events, + next_batch=SlidingSyncStickyEventsToken( + sticky_events_stream_id=sticky_events_to_id + ), + ) diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index 216ef3b071..836cee6c20 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/synapse/handlers/sliding_sync/room_lists.py @@ -52,6 +52,7 @@ from synapse.storage.roommember import ( RoomsForUserStateReset, ) from synapse.types import ( + Absent, MutableStateMap, RoomStreamToken, StateMap, @@ -71,7 +72,6 @@ from synapse.types.handlers.sliding_sync import ( from synapse.types.state import StateFilter from synapse.util import MutableOverlayMapping from synapse.util.duration import Duration -from synapse.util.sentinel import Sentinel if TYPE_CHECKING: from synapse.server import HomeServer @@ -112,31 +112,55 @@ class SlidingSyncInterestedRooms: sliding sync request. Returned by `compute_interested_rooms`. - - Attributes: - lists: A mapping from list name to the list result for the response - relevant_room_map: A map from rooms that match the sync request to - their room sync config. - relevant_rooms_to_send_map: Subset of `relevant_room_map` that - includes the rooms that *may* have relevant updates. Rooms not - in this map will definitely not have room updates (though - extensions may have updates in these rooms). - newly_joined_rooms: The set of rooms that were joined in the token range - and the user is still joined to at the end of this range. - newly_left_rooms: The set of rooms that we left in the token range - and are still "leave" at the end of this range. - dm_room_ids: The set of rooms the user consider as direct-message (DM) rooms """ lists: Mapping[str, SlidingSyncResult.SlidingWindowList] + """ + A mapping from list name to the list result for the response + """ + relevant_room_map: Mapping[str, RoomSyncConfig] + """ + A map from rooms that match the sync request to + their room sync config. + """ + relevant_rooms_to_send_map: Mapping[str, RoomSyncConfig] + """ + Subset of `relevant_room_map` that + includes the rooms that *may* have relevant updates. Rooms not + in this map will definitely not have room updates (though + extensions may have updates in these rooms). + """ + all_rooms: set[str] + """ + The set of room IDs of all rooms that could appear in any list. + This set includes rooms that are outside the list ranges. + In other words, this is the set of all rooms that the client is + _interested_ in (in a pure sense), + even if these rooms are omitted from the current window (which + is, in a sense, just a computational optimisation). + """ + room_membership_for_user_map: Mapping[str, RoomsForUserType] newly_joined_rooms: AbstractSet[str] + """ + The set of rooms that were joined in the token range + and the user is still joined to at the end of this range. + """ + newly_left_rooms: AbstractSet[str] + """ + The set of rooms that we left in the token range + and are still "leave" at the end of this range. + """ + dm_room_ids: AbstractSet[str] + """ + The set of rooms the user consider as direct-message (DM) rooms + """ @staticmethod def empty() -> "SlidingSyncInterestedRooms": @@ -1703,10 +1727,8 @@ class SlidingSyncRoomLists: # (applies to invite/knock rooms) rooms_ids_without_stripped_state: set[str] = set() for room_id in room_ids_without_results: - stripped_state_map = room_id_to_stripped_state_map.get( - room_id, Sentinel.UNSET_SENTINEL - ) - assert stripped_state_map is not Sentinel.UNSET_SENTINEL, ( + stripped_state_map = room_id_to_stripped_state_map.get(room_id, Absent) + assert stripped_state_map is not Absent, ( f"Stripped state left unset for room {room_id}. " + "Make sure you're calling `_bulk_get_stripped_state_for_rooms_from_sync_room_map(...)` " + "with that room_id. (this is a problem with Synapse itself)" diff --git a/synapse/module_api/__init__.py b/synapse/module_api/__init__.py index 947be24d3e..3341e49b85 100644 --- a/synapse/module_api/__init__.py +++ b/synapse/module_api/__init__.py @@ -142,6 +142,8 @@ from synapse.storage.background_updates import ( from synapse.storage.database import DatabasePool, LoggingTransaction from synapse.storage.databases.main.roommember import ProfileInfo from synapse.types import ( + Absent, + AbsentType, DomainSpecificString, JsonDict, JsonMapping, @@ -160,7 +162,6 @@ from synapse.util.caches.descriptors import CachedFunction, cached as _cached from synapse.util.clock import Clock from synapse.util.duration import Duration from synapse.util.frozenutils import freeze -from synapse.util.sentinel import Sentinel if TYPE_CHECKING: # Old versions don't have `LiteralString` @@ -1990,7 +1991,7 @@ class ModuleApi: self, user_id: UserID, new_displayname: str, - deactivation: bool | Sentinel = Sentinel.UNSET_SENTINEL, + deactivation: bool | AbsentType = Absent, ) -> None: """Sets a user's display name. @@ -2020,7 +2021,7 @@ class ModuleApi: """ requester = create_requester(user_id) - if deactivation is not Sentinel.UNSET_SENTINEL: + if deactivation is not Absent: logger.error( "Deprecated `deactivation` parameter passed to `set_displayname` Module API (value: %r). This will break in 2027.", deactivation, diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py index c3cf0dc3c4..702ddcd6ca 100644 --- a/synapse/rest/client/sync.py +++ b/synapse/rest/client/sync.py @@ -20,7 +20,7 @@ # import logging from collections import defaultdict -from typing import TYPE_CHECKING, Any, Mapping +from typing import TYPE_CHECKING, Any, Literal, Mapping import attr @@ -672,6 +672,7 @@ class SlidingSyncRestServlet(RestServlet): - receipts (MSC3960) - account data (MSC3959) - thread subscriptions (MSC4308) + - sticky events (MSC4354) Request query parameters: timeout: How long to wait for new events in milliseconds. @@ -895,7 +896,7 @@ class SlidingSyncRestServlet(RestServlet): requester, sliding_sync_result.rooms ) response["extensions"] = await self.encode_extensions( - requester, sliding_sync_result.extensions + requester, sliding_sync_result.extensions, sliding_sync_result.rooms ) return response @@ -1045,8 +1046,18 @@ class SlidingSyncRestServlet(RestServlet): @trace_with_opname("sliding_sync.encode_extensions") async def encode_extensions( - self, requester: Requester, extensions: SlidingSyncResult.Extensions + self, + requester: Requester, + extensions: SlidingSyncResult.Extensions, + ref_rooms_results: Mapping[str, SlidingSyncResult.RoomResult], ) -> JsonDict: + """ + Args: + ref_rooms_results: + Map of room ID -> RoomResult that was serialised as the `room` section + of the Sliding Sync response. + Will not be mutated, only used for reading. + """ serialized_extensions: JsonDict = {} if extensions.to_device is not None: @@ -1115,8 +1126,80 @@ class SlidingSyncRestServlet(RestServlet): _serialise_thread_subscriptions(extensions.thread_subscriptions) ) + if extensions.sticky_events: + serialized_extensions[ + "org.matrix.msc4354.sticky_events" + ] = await self._serialise_sticky_events( + requester, extensions.sticky_events, ref_rooms_results + ) + return serialized_extensions + async def _serialise_sticky_events( + self, + requester: Requester, + sticky_events: SlidingSyncResult.Extensions.StickyEventsExtension, + ref_rooms_results: Mapping[str, SlidingSyncResult.RoomResult], + ) -> JsonDict: + """ + Serialise the sticky events extension response. + + This includes deduplicating by filtering out sticky events + from this extension that already appeared in the timeline + section. + + Args: + ref_rooms_results: + Map of room ID -> RoomResult that was serialised as the `room` section + of the Sliding Sync response. + Will not be mutated, only used for reading. + """ + + time_now = self.clock.time_msec() + # Same as SSS timelines. + # + serialize_options = SerializeEventConfig( + event_format=format_event_for_client_v2_without_room_id, + requester=requester, + ) + + rooms_out: dict[str, dict[Literal["events"], list[JsonDict]]] = {} + for ( + room_id, + possibly_duplicated_sticky_events, + ) in sticky_events.room_id_to_sticky_events.items(): + # As per MSC4354: + # Remove sticky events that are already in the timeline, else we will needlessly duplicate + # events. + # There is no purpose in including sticky events in the sticky section if they're already in + # the timeline, as either way the client becomes aware of them. + # This is particularly important given the risk of sticky events spam since + # anyone can send sticky events, so halving the bandwidth on average for each sticky + # event is helpful. + room_result = ref_rooms_results.get(room_id) + if room_result is None: + # Nothing to deduplicate + sticky_events_to_write = possibly_duplicated_sticky_events + else: + sent_event_ids_in_room_section = { + ev.event.event_id for ev in room_result.timeline_events + } + sticky_events_to_write = [ + ev + for ev in possibly_duplicated_sticky_events + if ev.event.event_id not in sent_event_ids_in_room_section + ] + rooms_out[room_id] = { + "events": await self.event_serializer.serialize_events( + sticky_events_to_write, time_now, config=serialize_options + ) + } + + return { + "rooms": rooms_out, + "next_batch": sticky_events.next_batch.serialise(), + } + def _serialise_thread_subscriptions( thread_subscriptions: SlidingSyncResult.Extensions.ThreadSubscriptionsExtension, diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index a66caa672c..9c5fd35906 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -1509,6 +1509,43 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas "last_receipt_stream_ordering": stream_ordering, }, ) + # If no summary row exists yet for a thread that has pending push + # actions (room active but not yet through a rotation cycle), the + # UPDATE above is a silent no-op for that thread and + # last_receipt_stream_ordering is never persisted. + # _rotate_notifs_before_txn would then INSERT the row with + # last_receipt_stream_ordering=NULL, causing the badge query to + # include every event before the receipt as unread. Pre-populate + # rows for every thread with pending push actions so rotation + # only counts events that arrive after this receipt. + txn.execute( + """ + SELECT DISTINCT thread_id + FROM event_push_actions + WHERE user_id = ? AND room_id = ? + """, + (user_id, room_id), + ) + pending_thread_ids = [row[0] for row in txn] + self.db_pool.simple_upsert_many_txn( + txn, + table="event_push_summary", + key_names=("user_id", "room_id", "thread_id"), + key_values=[ + (user_id, room_id, pending_thread_id) + for pending_thread_id in pending_thread_ids + ], + value_names=( + "notif_count", + "unread_count", + "stream_ordering", + "last_receipt_stream_ordering", + ), + value_values=[ + (0, 0, old_rotate_stream_ordering, stream_ordering) + for _ in pending_thread_ids + ], + ) # For a threaded receipt, we *always* want to update that receipt, # event if there are no new notifications in that thread. This ensures @@ -1517,8 +1554,10 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas unread_counts = [(0, 0, thread_id)] # Then any updated threads get their notification count and unread - # count updated. - self.db_pool.simple_update_many_txn( + # count updated. Use upsert so that a row is created if none exists + # yet (same race as the unthreaded case above: without this, rotation + # would INSERT with last_receipt_stream_ordering=NULL). + self.db_pool.simple_upsert_many_txn( txn, table="event_push_summary", key_names=("room_id", "user_id", "thread_id"), diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py index f2b1c3e42b..a6fc806701 100644 --- a/synapse/types/__init__.py +++ b/synapse/types/__init__.py @@ -27,8 +27,10 @@ from enum import Enum from typing import ( TYPE_CHECKING, AbstractSet, + Annotated, Any, ClassVar, + Final, Literal, Mapping, Match, @@ -41,8 +43,12 @@ from typing import ( overload, ) +import annotated_types import attr +import pydantic_core.core_schema from immutabledict import immutabledict +from pydantic import GetCoreSchemaHandler, StrictInt +from pydantic_core import CoreSchema from signedjson.key import decode_verify_key_bytes from signedjson.types import VerifyKey from typing_extensions import Self @@ -109,6 +115,110 @@ StrCollection = tuple[str, ...] | list[str] | AbstractSet[str] StrSequence = tuple[str, ...] | list[str] +class AbsentType(Enum): + """ + Type of a sentinel to use as an alternative to `None` + for when we really mean 'absent' and not JSON null. + + Generally suitable for distinguishing a default state from user-suppliable values. + Has no meaning on its own. + + It is falsy (like None is), so shorthand forms like `x or 0` can be used. + """ + + # Making this an Enum member makes this compatible with type narrowing, + # meaning `x is not Absent` will narrow `x: int | AbsentType` to `x: int` etc. + _Absent = object() + + @classmethod + def __get_pydantic_core_schema__( + cls, source_type: object, handler: GetCoreSchemaHandler + ) -> CoreSchema: + """ + This function is checked for and used by Pydantic when + attempting to deserialise/validate a field of this type. + + As the `Absent` type has no valid value when deserialising + from JSON (as that's the point; `Absent` is a marker representing + a lack of any JSON value), we always reject any value. + Instead of deserialising from this type, we rely on the struct class + we are in having field defaults that provide an `Absent`, which does not + go through the JSON validation. + + When validating Python, we accept the absent marker itself. + """ + + def _reject_from_json(v: object) -> "AbsentType": + """ + Reject the JSON value, no matter what it is, since absent values + are meant to be ... absent, thus have nothing they can be deserialised + from. + """ + raise ValueError("AbsentType cannot be deserialized from JSON") + + # `json_or_python_schema` wrapper needed for Pydantic < 2.10 + # but can be replaced with just the `is_instance_schema` after that version. + return pydantic_core.core_schema.json_or_python_schema( + json_schema=pydantic_core.core_schema.no_info_plain_validator_function( + _reject_from_json + ), + python_schema=pydantic_core.core_schema.is_instance_schema(cls), + ) + + def __copy__(self) -> "AbsentType": + """ + Copy implementation used by `copy.copy()`. + Always use the same instance. + + Without this and the deep version `__deepcopy__`, + `copy.copy(Absent)` on Python 3.10 (olddeps) + had a problem where it tried to construct a new Absent + as part of a deepcopy operation and resulted in: + ValueError: is not a valid AbsentType + """ + return self + + def __deepcopy__(self, memo: object) -> "AbsentType": + """ + Copy implementation used by `copy.deepcopy()`. + Always use the same instance. + """ + return self + + def __bool__(self) -> Literal[False]: + return False + + def __str__(self) -> str: + return "Absent" + + def __repr__(self) -> str: + return "Absent" + + +Absent: Final = AbsentType._Absent +""" +Sentinel to use as an alternative to `None` +for when we really mean 'absent' and not JSON null. + +Generally suitable for distinguishing a default state from user-suppliable values. +Has no meaning on its own. + +It is falsy (like None is), so shorthand forms like `x or 0` can be used. + +(Previously known as `Sentinel.UNSET_SENTINEL`.) +""" + + +NonNegativeStrictInt = Annotated[StrictInt, annotated_types.Ge(0)] +"""A strict integer that must be greater than or equal to zero. + +Should be preferred in place of Pydantic's own (lax) NonNegativeInt, +which will coerce strings to integers in a way that does not agree with +the Matrix specification (and would risk backing us into a backward compatibility +hole where we had to support input forms we didn't intend). +""" + + # Note that this seems to require inheriting *directly* from Interface in order # for mypy-zope to realize it is an interface. class ISynapseThreadlessReactor( diff --git a/synapse/types/handlers/sliding_sync.py b/synapse/types/handlers/sliding_sync.py index e73533d296..dd913250ba 100644 --- a/synapse/types/handlers/sliding_sync.py +++ b/synapse/types/handlers/sliding_sync.py @@ -48,7 +48,7 @@ from synapse.types import ( ThreadSubscriptionsToken, UserID, ) -from synapse.types.rest.client import SlidingSyncBody +from synapse.types.rest.client import SlidingSyncBody, SlidingSyncStickyEventsToken from synapse.util.clock import Clock from synapse.util.duration import Duration @@ -424,12 +424,31 @@ class SlidingSyncResult: or bool(self.prev_batch) ) + @attr.s(slots=True, frozen=True, auto_attribs=True) + class StickyEventsExtension: + """The Sticky Events extension (MSC4354) + + Attributes: + room_id_to_sticky_events: map (room_id -> [unexpired_sticky_events]) + The events are ordered by the sticky events stream. + + The events haven't yet been deduplicated to remove + events that also appear in the timeline. + """ + + room_id_to_sticky_events: Mapping[str, list[FilteredEvent]] + next_batch: SlidingSyncStickyEventsToken + + def __bool__(self) -> bool: + return bool(self.room_id_to_sticky_events) + to_device: ToDeviceExtension | None = None e2ee: E2eeExtension | None = None account_data: AccountDataExtension | None = None receipts: ReceiptsExtension | None = None typing: TypingExtension | None = None thread_subscriptions: ThreadSubscriptionsExtension | None = None + sticky_events: StickyEventsExtension | None = None def __bool__(self) -> bool: """Are there any updates that should be returned immediately to @@ -441,6 +460,7 @@ class SlidingSyncResult: or self.receipts or self.typing or self.thread_subscriptions + or self.sticky_events ) next_pos: SlidingSyncStreamToken diff --git a/synapse/types/rest/client/__init__.py b/synapse/types/rest/client/__init__.py index 49782b5234..a7cb4d8b08 100644 --- a/synapse/types/rest/client/__init__.py +++ b/synapse/types/rest/client/__init__.py @@ -18,9 +18,14 @@ # [This file includes modifications made by New Vector Limited] # # +import re +from typing import ClassVar + +import pydantic_core.core_schema from pydantic import ( ConfigDict, Field, + GetCoreSchemaHandler, StrictBool, StrictInt, StrictStr, @@ -28,9 +33,10 @@ from pydantic import ( field_validator, model_validator, ) -from pydantic_core import PydanticCustomError +from pydantic_core import CoreSchema, PydanticCustomError from typing_extensions import Annotated, Self +from synapse.types import Absent, AbsentType, NonNegativeStrictInt from synapse.types.rest import RequestBodyModel from synapse.util.threepids import validate_email @@ -107,6 +113,82 @@ class MsisdnRequestTokenBody(ThreepidRequestTokenBody): phone_number: StrictStr +class SlidingSyncStickyEventsToken: + """ + A token returned by `next_batch` of the MSC4354 Sticky Events extension to Sliding Sync + and then accepted as the `since` parameter in the requests of the same extension. + + Current format: + SlidingSyncStickyEventsToken ::= 'sticky_' DIGIT+ + DIGIT ::= '0'-'9' + + The `sticky_` prefix allows us to make sure it's not swapped for another token + or to evolve the type of token accepted with backwards compatibility in the future. + """ + + PATTERN = re.compile(r"^sticky_([0-9]+)$") + START: ClassVar["SlidingSyncStickyEventsToken"] + + def __init__(self, *, sticky_events_stream_id: int) -> None: + # FIXME: We should use MultiWriterStreamToken here + # Track: https://github.com/element-hq/synapse/issues/19661 + self.sticky_events_stream_id = sticky_events_stream_id + + @classmethod + def __get_pydantic_core_schema__( + cls, source_type: object, handler: GetCoreSchemaHandler + ) -> CoreSchema: + """ + This function is checked for and used by Pydantic when + attempting to deserialise/validate a field of this type. + + This returns a schema that will parse a string into an + instance of `SlidingSyncStickyEventsToken`. + """ + + return pydantic_core.core_schema.no_info_plain_validator_function( + cls._validate, + serialization=pydantic_core.core_schema.plain_serializer_function_ser_schema( + cls.serialise, + info_arg=False, + ), + ) + + @classmethod + def _validate(cls, v: object) -> Self: + """ + Create an instance from serialised string form. + + The inverse of `serialise`. + """ + if isinstance(v, cls): + return v + if isinstance(v, str): + match = cls.PATTERN.match(v) + if match is None: + raise ValueError(f"Invalid SlidingSyncStickyEventsToken format: {v!r}") + return cls(sticky_events_stream_id=int(match.group(1))) + raise ValueError(f"Cannot parse SlidingSyncStickyEventsToken from {type(v)}") + + def serialise(self) -> str: + """ + Convert this instance to string. + + The inverse of `_validate`. + """ + return f"sticky_{self.sticky_events_stream_id}" + + def __repr__(self) -> str: + # Use the serialised form as debug output. + return self.serialise() + + +# Starting reading a stream at 0 ensures all stream fact rows will be read +SlidingSyncStickyEventsToken.START = SlidingSyncStickyEventsToken( + sticky_events_stream_id=0 +) + + class SlidingSyncBody(RequestBodyModel): """ Sliding Sync API request body. @@ -383,6 +465,19 @@ class SlidingSyncBody(RequestBodyModel): enabled: StrictBool | None = False limit: StrictInt = 100 + class StickyEventsExtension(RequestBodyModel): + """The Sticky Events extension (MSC4354) + + Attributes: + enabled + limit: maximum number of sticky events to return in the extension (default 100) + since: either a string with the Sticky Events since token or absent + """ + + enabled: StrictBool = False + limit: NonNegativeStrictInt = 100 + since: SlidingSyncStickyEventsToken | AbsentType = Absent + to_device: ToDeviceExtension | None = None e2ee: E2eeExtension | None = None account_data: AccountDataExtension | None = None @@ -391,6 +486,9 @@ class SlidingSyncBody(RequestBodyModel): thread_subscriptions: ThreadSubscriptionsExtension | None = Field( None, alias="io.element.msc4308.thread_subscriptions" ) + sticky_events: StickyEventsExtension | AbsentType = Field( + Absent, alias="org.matrix.msc4354.sticky_events" + ) conn_id: StrictStr | None = None lists: ( diff --git a/synapse/util/async_helpers.py b/synapse/util/async_helpers.py index 120c12f048..5211bcc8e1 100644 --- a/synapse/util/async_helpers.py +++ b/synapse/util/async_helpers.py @@ -390,6 +390,7 @@ T3 = TypeVar("T3") T4 = TypeVar("T4") T5 = TypeVar("T5") T6 = TypeVar("T6") +T7 = TypeVar("T7") @overload @@ -519,6 +520,30 @@ async def gather_optional_coroutines( ) -> tuple[T1 | None, T2 | None, T3 | None, T4 | None, T5 | None, T6 | None]: ... +@overload +async def gather_optional_coroutines( + *coroutines: Unpack[ + tuple[ + Coroutine[Any, Any, T1] | None, + Coroutine[Any, Any, T2] | None, + Coroutine[Any, Any, T3] | None, + Coroutine[Any, Any, T4] | None, + Coroutine[Any, Any, T5] | None, + Coroutine[Any, Any, T6] | None, + Coroutine[Any, Any, T7] | None, + ] + ], +) -> tuple[ + T1 | None, + T2 | None, + T3 | None, + T4 | None, + T5 | None, + T6 | None, + T7 | None, +]: ... + + async def gather_optional_coroutines( *coroutines: Unpack[tuple[Coroutine[Any, Any, T1] | None, ...]], ) -> tuple[T1 | None, ...]: diff --git a/synapse/util/sentinel.py b/synapse/util/sentinel.py deleted file mode 100644 index c8434fc97a..0000000000 --- a/synapse/util/sentinel.py +++ /dev/null @@ -1,21 +0,0 @@ -# -# This file is licensed under the Affero General Public License (AGPL) version 3. -# -# Copyright (C) 2025 New Vector, Ltd -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. -# -# See the GNU Affero General Public License for more details: -# . -# - -import enum - - -class Sentinel(enum.Enum): - # defining a sentinel in this way allows mypy to correctly handle the - # type of a dictionary lookup and subsequent type narrowing. - UNSET_SENTINEL = object() diff --git a/synapse/visibility.py b/synapse/visibility.py index fc3e9dfa49..4eeba14a6c 100644 --- a/synapse/visibility.py +++ b/synapse/visibility.py @@ -103,6 +103,7 @@ async def filter_and_transform_events_for_client( Returns: The filtered events, wrapped in FilteredEvent with the requesting user's membership at each event annotated for use during serialization (MSC4115). + The events are returned in the same order. """ # Filter out events that have been soft failed so that we don't relay them # to clients, unless they're a server admin and want that to happen. diff --git a/tests/rest/client/sliding_sync/test_extension_sticky_events.py b/tests/rest/client/sliding_sync/test_extension_sticky_events.py new file mode 100644 index 0000000000..aaea6d6793 --- /dev/null +++ b/tests/rest/client/sliding_sync/test_extension_sticky_events.py @@ -0,0 +1,676 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 New Vector, Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . +# +import logging +import sqlite3 + +from twisted.internet.testing import MemoryReactor + +import synapse.rest.admin +import synapse.rest.client.account_data +from synapse.api.constants import EventTypes, EventUnsignedContentFields +from synapse.rest.client import account_data, login, register, room, sync +from synapse.server import HomeServer +from synapse.types import JsonDict, StreamKeyType +from synapse.util.clock import Clock +from synapse.util.duration import Duration + +from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase +from tests.server import TimedOutException +from tests.utils import USE_POSTGRES_FOR_TESTS + +logger = logging.getLogger(__name__) + + +DUMMY_LISTS = { + "main": { + # Don't include any rooms in the top-N window + "ranges": [[0, 0]], + "required_state": [], + "timeline_limit": 0, + } +} +""" +Subscription lists that can be used in the Sliding Sync request `lists` field, +which sets up a subscription that is interested in all rooms but does not let any rooms into the window, +thus does not return any timelines. + +Sufficient to get sticky event updates as per MSC4354: + +> The server MUST include sticky events across all rooms that would be matched by at least one subscription list +> (i.e. all rooms that the client is interested in), even if the room does not appear in top-N window for that +> subscription list at this time. +> Rooms that would not be matched by a list are not included, as this means the client is not interested +> in those rooms. +> +> — https://github.com/matrix-org/matrix-spec-proposals/pull/4354/changes#diff-d76bc1a1d612c6da37d024f5b57f7b8352939b8db8a7ee9c6b71c1a848359afdR213-R217 +""" + + +class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): + """Tests for the sticky events sliding sync extension""" + + if not USE_POSTGRES_FOR_TESTS and sqlite3.sqlite_version_info < (3, 40, 0): + # We need the JSON functionality in SQLite + skip = f"SQLite version is too old to support sticky events: {sqlite3.sqlite_version_info} (See https://github.com/element-hq/synapse/issues/19428)" + + servlets = [ + synapse.rest.admin.register_servlets, + login.register_servlets, + register.register_servlets, + room.register_servlets, + sync.register_servlets, + account_data.register_servlets, + ] + + def default_config(self) -> JsonDict: + config = super().default_config() + # Enable sliding sync and sticky events MSCs + config["experimental_features"] = { + "msc3575_enabled": True, + "msc4354_enabled": True, + } + return config + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = hs.get_datastores().main + super().prepare(reactor, clock, hs) + + def _assert_sticky_events_response( + self, + response_body: JsonDict, + expected_events_by_room: dict[str, list[str]] | None, + ) -> str | None: + """Assert the sliding sync response was successful and has the expected + sticky events. + + Args: + response_body: Sliding Sync response body + expected_events_by_room: + map of room ID to list of event IDs to expect (in the order we expect them), + or None if we expect an empty sticky events extension response + + Returns the next_batch token from the sticky events section, + unless we're expecting an empty response. + """ + extensions = response_body["extensions"] + sticky_events = extensions.get("org.matrix.msc4354.sticky_events") + + # If there are no expected events, we shouldn't get anything in the response + if expected_events_by_room is None: + self.assertIsNone(sticky_events) + return None + + self.assertIsNotNone(sticky_events) + self.assertIsInstance(sticky_events["next_batch"], str) + + actual_rooms = sticky_events["rooms"] + # Check that we have the expected rooms + self.assertIncludes( + set(actual_rooms.keys()), set(expected_events_by_room.keys()), exact=True + ) + + # Check the events in each room + for room_id, expected_events in expected_events_by_room.items(): + actual_events = actual_rooms[room_id]["events"] + actual_event_ids = [e["event_id"] for e in actual_events] + self.assertEqual(actual_event_ids, expected_events) + for actual_event in actual_events: + # Check the sticky TTL is sent + self.assertIn("unsigned", actual_event) + ttl = actual_event["unsigned"][EventUnsignedContentFields.STICKY_TTL] + self.assertIsInstance(ttl, int) + + self.assertIn("next_batch", sticky_events) + return sticky_events["next_batch"] + + def test_empty_sync(self) -> None: + """Test that enabling sticky events extension works on initial and incremental sync, + even if there is no data. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + + sync_body = { + "lists": DUMMY_LISTS, + "extensions": { + "org.matrix.msc4354.sticky_events": { + "enabled": True, + } + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + # No sticky events in initial sync. + self._assert_sticky_events_response(response_body, None) + + # Incremental sync should also have no sticky events + response_body, _ = self.do_sync( + sync_body, since=response_body["pos"], tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + + def test_initial_sync(self) -> None: + """Test that we get sticky events when we don't specify a since token + (initial sync). + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("u2", "pass") + user2_tok = self.login(user2_id, "pass") + + # Create a room and join both users + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + # Send a sticky event from user2 + sticky_event_id: str = self.helper.send_sticky_event( + room_id, + EventTypes.Message, + duration=Duration(minutes=5), + content={"body": "sticky message", "msgtype": "m.text"}, + tok=user2_tok, + )["event_id"] + + # Initial sync should return the sticky event + sync_body: JsonDict = { + "lists": DUMMY_LISTS, + "extensions": { + "org.matrix.msc4354.sticky_events": { + "enabled": True, + } + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + + # Assert the response and then get the next_batch for the next sliding sync request + next_batch = self._assert_sticky_events_response( + response_body, {room_id: [sticky_event_id]} + ) + assert next_batch is not None + + # Do an incremental sync immediately again + sync_body = { + "lists": DUMMY_LISTS, + "extensions": { + "org.matrix.msc4354.sticky_events": { + "enabled": True, + "since": next_batch, + } + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + + # Check we don't get that event again + self._assert_sticky_events_response(response_body, None) + + # Send another sticky event + sticky_event_id2: str = self.helper.send_sticky_event( + room_id, + EventTypes.Message, + duration=Duration(minutes=5), + content={"body": "another sticky message", "msgtype": "m.text"}, + tok=user1_tok, + )["event_id"] + + # Now the incremental sync should give us that event + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + self._assert_sticky_events_response( + response_body, {room_id: [sticky_event_id2]} + ) + + def test_expired_events_not_returned(self) -> None: + """Test that expired sticky events are not returned.""" + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("u2", "pass") + user2_tok = self.login(user2_id, "pass") + + # Create a room + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + # Send a sticky event with a short duration + sticky_event_id = self.helper.send_sticky_event( + room_id, + EventTypes.Message, + duration=Duration(seconds=2), + content={"body": "sticky message", "msgtype": "m.text"}, + tok=user2_tok, + )["event_id"] + + # Initial sync should return the sticky event + sync_body = { + "lists": DUMMY_LISTS, + "extensions": { + "org.matrix.msc4354.sticky_events": { + "enabled": True, + } + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + + # We should still get the event for now + self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]}) + + # Advance time past the sticky duration + self.reactor.advance(3) + + # A second initial sync should not return the expired sticky event + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + self._assert_sticky_events_response(response_body, None) + + def test_wait_for_new_data(self) -> None: + """Test that the sliding sync request waits for new sticky events to arrive. + (Only applies to incremental syncs with a `timeout` specified). + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("u2", "pass") + user2_tok = self.login(user2_id, "pass") + + # Create a room + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + # Initial sync with no sticky events + sync_body = { + "lists": DUMMY_LISTS, + "extensions": { + "org.matrix.msc4354.sticky_events": { + "enabled": True, + } + }, + } + _, from_token = self.do_sync(sync_body, tok=user1_tok) + + # Make the sliding sync request with a timeout + channel = self.make_request( + "POST", + self.sync_endpoint + "?timeout=10000" + f"&pos={from_token}", + content=sync_body, + access_token=user1_tok, + await_result=False, + ) + + # Block for 5 seconds to make sure we are in `notifier.wait_for_events(...)` + with self.assertRaises(TimedOutException): + channel.await_result(timeout_ms=5000) + + # Send a sticky event to trigger new results + sticky_event_id = self.helper.send_sticky_event( + room_id, + EventTypes.Message, + duration=Duration(minutes=5), + content={"body": "sticky message", "msgtype": "m.text"}, + tok=user2_tok, + )["event_id"] + + # Should respond before the 10 second timeout + channel.await_result(timeout_ms=100) + self.assertEqual(channel.code, 200, channel.json_body) + + self._assert_sticky_events_response( + channel.json_body, + {room_id: [sticky_event_id]}, + ) + + def test_wait_for_new_data_timeout(self) -> None: + """ + Test that the sliding sync request waits for new sticky events to arrive + and times out when no data arrives before the deadline. + (Only applies to incremental syncs with a `timeout` specified). + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("u2", "pass") + user2_tok = self.login(user2_id, "pass") + + # Create a room + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + # Initial sync with no sticky events + sync_body = { + "lists": DUMMY_LISTS, + "extensions": { + "org.matrix.msc4354.sticky_events": { + "enabled": True, + } + }, + } + _, from_token = self.do_sync(sync_body, tok=user1_tok) + + # Make the sliding sync request with a timeout + channel = self.make_request( + "POST", + self.sync_endpoint + "?timeout=10000" + f"&pos={from_token}", + content=sync_body, + access_token=user1_tok, + await_result=False, + ) + + # Block for 5 seconds to make sure we are `notifier.wait_for_events(...)` + with self.assertRaises(TimedOutException): + channel.await_result(timeout_ms=5000) + # Wake-up `notifier.wait_for_events(...)` that will cause us test + # `SlidingSyncResult.__bool__` for new results. + self._bump_notifier_wait_for_events( + # wake key is intentionally unrelated to sticky events + user1_id, + wake_stream_key=StreamKeyType.ACCOUNT_DATA, + ) + # Block for a little bit more to ensure we don't see any new results. + with self.assertRaises(TimedOutException): + channel.await_result(timeout_ms=4000) + # Wait for the sync to complete (wait for the rest of the 10 second timeout, + # 5000 + 4000 + 1200 > 10000) + channel.await_result(timeout_ms=1200) + self.assertEqual(channel.code, 200, channel.json_body) + + self._assert_sticky_events_response( + channel.json_body, + None, + ) + + def test_ignored_users_sticky_events(self) -> None: + """ + Test that sticky events from ignored users are not delivered to clients. + + > As with normal events, sticky events sent by ignored users MUST NOT be + > delivered to clients. + > — https://github.com/matrix-org/matrix-spec-proposals/blob/4340903c15e9eab1bfb2f6a31cfa08fd535f7e7c/proposals/4354-sticky-events.md#sync-api-changes + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + # Create a room + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + # User1 ignores user2 + channel = self.make_request( + "PUT", + f"/_matrix/client/v3/user/{user1_id}/account_data/m.ignored_user_list", + {"ignored_users": {user2_id: {}}}, + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) + + # User2 sends a sticky event + sticky_event_id = self.helper.send_sticky_event( + room_id, + EventTypes.Message, + duration=Duration(minutes=5), + content={"body": "sticky from ignored user", "msgtype": "m.text"}, + tok=user2_tok, + )["event_id"] + + # Initial sync for user1 + sync_body = { + "lists": { + "main": { + "ranges": [[0, 10]], + "required_state": [], + # In this test we ask for 10 events of timeline. + "timeline_limit": 10, + } + }, + "extensions": { + "org.matrix.msc4354.sticky_events": { + "enabled": True, + } + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + + # Timeline events should not include sticky event from ignored user + timeline_events = response_body["rooms"][room_id]["timeline"] + timeline_event_ids = [e["event_id"] for e in timeline_events] + + self.assertNotIn( + sticky_event_id, + timeline_event_ids, + "Sticky event from ignored user should not be in timeline", + ) + + # Sticky events section should also not include the event from ignored user + self._assert_sticky_events_response(response_body, None) + + def test_history_visibility_bypass_for_sticky_events(self) -> None: + """ + Test that joined users can see sticky events even when history visibility + is set to "joined" and they joined after the event was sent. + + > History visibility checks MUST NOT be applied to sticky events. + > Any joined user is authorised to see sticky events for the duration they remain sticky. + > — https://github.com/matrix-org/matrix-spec-proposals/blob/4340903c15e9eab1bfb2f6a31cfa08fd535f7e7c/proposals/4354-sticky-events.md#proposal + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + + # Create a room with restrictive history visibility + room_id = self.helper.create_room_as( + user1_id, + tok=user1_tok, + extra_content={ + # Anyone can join + "preset": "public_chat", + # But you can't see history before you joined + "initial_state": [ + { + "type": EventTypes.RoomHistoryVisibility, + "state_key": "", + "content": {"history_visibility": "joined"}, + } + ], + }, + is_public=False, + ) + + # User1 sends a sticky event + sticky_event_id = self.helper.send_sticky_event( + room_id, + EventTypes.Message, + duration=Duration(minutes=5), + content={"body": "sticky message", "msgtype": "m.text"}, + tok=user1_tok, + )["event_id"] + + # User1 also sends a regular event, to verify our test setup + regular_event_id = self.helper.send( + room_id=room_id, + body="regular message", + tok=user1_tok, + )["event_id"] + + # Register and join a second user after the sticky event was sent + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + self.helper.join(room_id, user2_id, tok=user2_tok) + + # User2 syncs - they should see sticky event even though + # history visibility is "joined" and they joined after it was sent + sync_body = { + "lists": { + "main": { + "ranges": [[0, 10]], + "required_state": [], + # In this test, we ask for 10 events of timeline. + "timeline_limit": 10, + } + }, + "extensions": { + "org.matrix.msc4354.sticky_events": { + "enabled": True, + } + }, + } + response_body, _ = self.do_sync(sync_body, tok=user2_tok) + + # The sticky event is fully visible in its own right, + # but AFAICT the timeline only includes events since we join the room + # (regardless of history visibility), + # so this comes down in the sticky extension + self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]}) + + # Instead the sticky event is in the timeline + timeline_events = response_body["rooms"][room_id]["timeline"] + timeline_event_ids = [e["event_id"] for e in timeline_events] + self.assertNotIn( + regular_event_id, + timeline_event_ids, + f"Expecting to not see regular event ({regular_event_id}) before user1 joined.", + ) + + def test_sticky_event_pagination(self) -> None: + """ + Test that pagination works correctly when there are many sticky events. + Also check they are delivered in stream order. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + # Create a room + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + # Send 4 sticky events (more than our limit of 2) + sticky_event_ids: list[str] = [] + for i in range(4): + event_id = self.helper.send_sticky_event( + room_id, + EventTypes.Message, + duration=Duration(minutes=5), + content={"body": f"sticky message {i}", "msgtype": "m.text"}, + tok=user2_tok, + )["event_id"] + sticky_event_ids.append(event_id) + + # Initial sync + sync_body = { + "lists": DUMMY_LISTS, + "extensions": { + "org.matrix.msc4354.sticky_events": {"enabled": True, "limit": 2} + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + + # We expect to see the first 2 sticky events by stream order + # and they should be in that stream order + next_batch = self._assert_sticky_events_response( + response_body, {room_id: sticky_event_ids[0:2]} + ) + + # Incremental sync to get remaining sticky events + sync_body = { + "lists": DUMMY_LISTS, + "extensions": { + "org.matrix.msc4354.sticky_events": { + "enabled": True, + # This makes it incremental + "since": next_batch, + "limit": 3, + } + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + + # Should get remaining events, in stream order again + self._assert_sticky_events_response( + response_body, {room_id: sticky_event_ids[2:4]} + ) + + def test_deduplication_with_timeline(self) -> None: + """ + Test that sticky events are not included in the sticky event extension of sliding sync + if they are included in the main timeline section. + + Send 3 events: + 1. sticky + 2. sticky + 3. regular + + We then will sync with a timeline limit of 2 and a sticky event limit of 2. + We should then see (2) and (3) included in the timeline + and (1) in the sticky event response (but not (2) because it's already + included in the timeline.) + + 1. sticky [in sticky section] + + ------------->>> Timeline section + 2. sticky + 3. regular + -------------<<< + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + room_id = self.helper.create_room_as(user1_id, tok=user1_tok) + + sticky_event_ids: list[str] = [] + for i in range(2): + event_id = self.helper.send_sticky_event( + room_id, + EventTypes.Message, + duration=Duration(minutes=5), + content={"body": f"sticky message {i}", "msgtype": "m.text"}, + tok=user1_tok, + )["event_id"] + sticky_event_ids.append(event_id) + + non_sticky_event_id = self.helper.send_event( + room_id, + EventTypes.Message, + content={"body": "regular message", "msgtype": "m.text"}, + tok=user1_tok, + )["event_id"] + + # Sync + sync_body = { + "lists": { + "main": { + "ranges": [[0, 10]], + "required_state": [], + # In this test, we want a timeline window of the 2 latest messages + "timeline_limit": 2, + } + }, + "extensions": { + "org.matrix.msc4354.sticky_events": { + "enabled": True, + "limit": 2, + } + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + + events_in_sticky_section = response_body["extensions"][ + "org.matrix.msc4354.sticky_events" + ]["rooms"][room_id]["events"] + event_ids_in_sticky_section = [e["event_id"] for e in events_in_sticky_section] + + events_in_timeline_section = response_body["rooms"][room_id]["timeline"] + event_ids_in_timeline_section = [ + e["event_id"] for e in events_in_timeline_section + ] + + self.assertEqual( + event_ids_in_sticky_section, + [sticky_event_ids[0]], + ) + self.assertEqual( + event_ids_in_timeline_section, [sticky_event_ids[1], non_sticky_event_id] + ) diff --git a/tests/rest/client/sliding_sync/test_sliding_sync.py b/tests/rest/client/sliding_sync/test_sliding_sync.py index fc7d6a279c..2fd18f0e54 100644 --- a/tests/rest/client/sliding_sync/test_sliding_sync.py +++ b/tests/rest/client/sliding_sync/test_sliding_sync.py @@ -374,7 +374,7 @@ class SlidingSyncBase(unittest.HomeserverTestCase): user_id: The user ID to wake up the notifier for wake_stream_key: The stream key to wake up. This will create an actual new entity in that stream so it's best to choose one that won't affect the - Sliding Sync results you're testing for. In other words, if your testing + Sliding Sync results you're testing for. In other words, if you're testing account data, choose `StreamKeyType.PRESENCE` instead. We support two possible stream keys because you're probably testing one or the other so one is always a "safe" option. diff --git a/tests/storage/test_event_push_actions.py b/tests/storage/test_event_push_actions.py index 85ebf26128..bfe7049fb2 100644 --- a/tests/storage/test_event_push_actions.py +++ b/tests/storage/test_event_push_actions.py @@ -363,6 +363,115 @@ class EventPushActionsStoreTestCase(HomeserverTestCase): _mark_read(cutoff_event_id) _assert_count(0) + def test_count_aggregation_receipt_before_first_rotation(self) -> None: + """ + Regression test: reading a highlight before the first rotation must not + permanently inflate the badge count. + + Highlights survive the receipt-triggered DELETE (highlight=1), so if + last_receipt_stream_ordering is NULL when rotation creates the summary + row, the badge query re-counts them forever. + """ + user_id, token, _, other_token, room_id = self._create_users_and_room() + + def _assert_badge(expected: int) -> None: + counts = self.get_success( + self.store.db_pool.runInteraction( + "get-aggregate-unread-counts", + self.store._get_unread_counts_by_room_for_user_txn, + user_id, + ) + ) + self.assertEqual(counts.get(room_id, 0), expected) + + def _send(highlight: bool = False) -> str: + return self.helper.send_event( + room_id, + type="m.room.message", + content={"msgtype": "m.text", "body": user_id if highlight else "msg"}, + tok=other_token, + )["event_id"] + + def _read(event_id: str) -> None: + self.get_success( + self.store.insert_receipt( + room_id, + "m.read", + user_id=user_id, + event_ids=[event_id], + thread_id=None, + data={}, + ) + ) + + # Highlight arrives; user reads it before any rotation (no summary row exists). + _read(_send(highlight=True)) + # One new event after the receipt makes stream_ordering > max_clause true. + _send() + # Without the fix: badge = 2 (highlight re-counted). With fix: badge = 1. + self.get_success(self.store._rotate_notifs()) + _assert_badge(1) + + def test_count_aggregation_receipt_before_first_rotation_in_thread(self) -> None: + """ + Same regression as test_count_aggregation_receipt_before_first_rotation, + but for a highlight inside a thread cleared by an unthreaded receipt. + + The fix must pre-populate event_push_summary for every thread with + pending push actions, not just MAIN_TIMELINE. + """ + user_id, token, _, other_token, room_id = self._create_users_and_room() + + def _assert_badge(expected: int) -> None: + counts = self.get_success( + self.store.db_pool.runInteraction( + "get-aggregate-unread-counts", + self.store._get_unread_counts_by_room_for_user_txn, + user_id, + ) + ) + self.assertEqual(counts.get(room_id, 0), expected) + + def _send(thread_root: str | None = None, highlight: bool = False) -> str: + content: JsonDict = { + "msgtype": "m.text", + "body": user_id if highlight else "msg", + } + if thread_root is not None: + content["m.relates_to"] = { + "rel_type": RelationTypes.THREAD, + "event_id": thread_root, + } + return self.helper.send_event( + room_id, + type="m.room.message", + content=content, + tok=other_token, + )["event_id"] + + def _read(event_id: str) -> None: + self.get_success( + self.store.insert_receipt( + room_id, + "m.read", + user_id=user_id, + event_ids=[event_id], + thread_id=None, + data={}, + ) + ) + + # A thread root, then a highlight inside that thread. + thread_root = _send() + thread_highlight = _send(thread_root=thread_root, highlight=True) + # User reads everything with an unthreaded receipt before any rotation. + _read(thread_highlight) + # A new event in the same thread makes stream_ordering > max_clause true. + _send(thread_root=thread_root) + # Without the fix: badge = 2 (thread highlight re-counted). With: badge = 1. + self.get_success(self.store._rotate_notifs()) + _assert_badge(1) + def test_count_aggregation_threads(self) -> None: """ This is essentially the same test as test_count_aggregation, but adds diff --git a/tests/test_types.py b/tests/test_types.py index 1802f0fae3..43fd96d6f5 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -18,16 +18,20 @@ # [This file includes modifications made by New Vector Limited] # # - +import copy from unittest import skipUnless from immutabledict import immutabledict from parameterized import parameterized_class +from pydantic import BaseModel, PydanticInvalidForJsonSchema, ValidationError from synapse.api.errors import SynapseError from synapse.types import ( + Absent, + AbsentType, AbstractMultiWriterStreamToken, MultiWriterStreamToken, + NonNegativeStrictInt, RoomAlias, RoomStreamToken, UserID, @@ -199,3 +203,150 @@ class MultiWriterTokenTestCase(unittest.HomeserverTestCase): parsed_token = self.get_success(self.token_type.parse(store, "m5~")) self.assertEqual(parsed_token, self.token_type(stream=5)) + + +class AbsentTestCase(unittest.TestCase): + """ + Tests for the `Absent` utility, which is meant to be like `None` except + explicitly signalling absence rather than JSON null. + """ + + def test_cant_create_second_absent(self) -> None: + """ + Tests that we aren't allowed to instantiate a second `Absent`. + """ + with self.assertRaises(TypeError): + AbsentType() # type: ignore[call-arg] + + def test_is_falsy(self) -> None: + """ + Tests `Absent` is falsy and can therefore be used a bit like `None`. + """ + if Absent: + self.fail("Absent is truthy!") + + self.assertEqual(Absent or "something", "something") + + def test_pydantic_jsonschema(self) -> None: + """ + Tests that `Absent` can't be used to produce JSONSchema in Pydantic models. + + In the future, it may be useful to produce correct JSONSchema, but for now + I was mostly interested in making sure we don't produce weird/invalid JSONSchema. + """ + + class MyModel(BaseModel): + absent: AbsentType = Absent + + with self.assertRaises(PydanticInvalidForJsonSchema): + MyModel.model_json_schema() + + def test_pydantic_reject_null(self) -> None: + """ + Tests that `Absent` rejects `None` (JSON null) when used in Pydantic models. + """ + + class MyModel(BaseModel): + absent: AbsentType = Absent + + with self.assertRaises(ValidationError): + MyModel.model_validate({"absent": None}) + + with self.assertRaises(ValidationError): + MyModel.model_validate_json('{"absent": null}') + + def test_pydantic_accept_absence(self) -> None: + """ + Tests that `Absent` accepts the absence of a value when used in Pydantic models. + """ + + class MyModel(BaseModel): + absent: AbsentType = Absent + + self.assertEqual(MyModel.model_validate({}), MyModel(absent=Absent)) + self.assertEqual(MyModel.model_validate_json("{}"), MyModel(absent=Absent)) + + def test_copy(self) -> None: + """ + Tests that the `copy` module always uses the same instance of Absent. + """ + + class MyModel(BaseModel): + absent: AbsentType = Absent + + a = MyModel.model_validate({}) + b = copy.deepcopy(a) + + self.assertIs(copy.copy(Absent), Absent) + self.assertIs(a.absent, b.absent) + + +class NonNegativeStrictIntTestCase(unittest.TestCase): + """ + Tests for the `NonNegativeStrictInt` utility. + """ + + def test_pydantic_jsonschema(self) -> None: + """ + Tests that `NonNegativeStrictInt` produces sensible JSONSchema. + """ + + class MyModel(BaseModel): + limit: NonNegativeStrictInt = 100 + + self.assertEqual( + MyModel.model_json_schema(), + { + "properties": { + "limit": { + "default": 100, + "minimum": 0, + "title": "Limit", + "type": "integer", + } + }, + "title": "MyModel", + "type": "object", + }, + f"JSONSchema actually is:\n{MyModel.model_json_schema()!r}", + ) + + def test_pydantic_reject(self) -> None: + """ + Tests that `NonNegativeStrictInt` rejects negative numbers + and non-ints. + """ + + class MyModel(BaseModel): + limit: NonNegativeStrictInt = 100 + + with self.assertRaises(ValidationError): + MyModel.model_validate({"limit": -1}) + + with self.assertRaises(ValidationError): + MyModel.model_validate_json('{"limit": -1}') + + # StrictInt, so don't accept floats... + with self.assertRaises(ValidationError): + MyModel.model_validate({"limit": 1.5}) + + with self.assertRaises(ValidationError): + MyModel.model_validate_json('{"limit": 1.5}') + + # ...and don't accept stringy ints either. + with self.assertRaises(ValidationError): + MyModel.model_validate({"limit": "42"}) + + with self.assertRaises(ValidationError): + MyModel.model_validate_json('{"limit": "42"}') + + def test_pydantic_accept(self) -> None: + """ + Tests that `Absent` accepts the absence of a value when used in Pydantic models. + """ + + class MyModel(BaseModel): + limit: NonNegativeStrictInt = 100 + + self.assertEqual(MyModel.model_validate_json('{"limit": 0}'), MyModel(limit=0)) + self.assertEqual(MyModel.model_validate({"limit": 42}), MyModel(limit=42))