diff --git a/.gitignore b/.gitignore index 1e2937230c..bb204dcfe1 100644 --- a/.gitignore +++ b/.gitignore @@ -81,3 +81,4 @@ book/ # Don't include users' poetry configs /poetry.toml +/localtest/ diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index df7748b822..106e2b09ef 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -119,6 +119,14 @@ class SlidingSyncHandler: # effective limit varies between requests by design. self.expanded_timeline_on_limit_increase = True + # Whether to persist each room's request config (`timeline_limit` + + # `required_state`) in the per-connection state, in order to detect + # config changes between requests. Paginated sync turns this off: its + # `required_state` is immutable for the life of a connection and + # `timeline_limit` changes carry no special semantics, so there is + # nothing to diff against. + self.track_room_configs = True + async def wait_for_sync_for_user( self, requester: Requester, @@ -1487,7 +1495,7 @@ class SlidingSyncHandler: required_state_map=room_sync_required_state_map_to_persist, ) - else: + elif self.track_room_configs: new_connection_state.room_configs[room_id] = RoomSyncConfig( timeline_limit=room_sync_config.timeline_limit, required_state_map=room_sync_required_state_map_to_persist, diff --git a/synapse/handlers/sliding_sync/paginated.py b/synapse/handlers/sliding_sync/paginated.py index 9d86d90900..d671890d7e 100644 --- a/synapse/handlers/sliding_sync/paginated.py +++ b/synapse/handlers/sliding_sync/paginated.py @@ -31,7 +31,9 @@ extensions, the connection store and the notifier integration are shared. import logging from typing import TYPE_CHECKING +from synapse.api.errors import SlidingSyncUnknownPosition from synapse.handlers.sliding_sync import SlidingSyncHandler +from synapse.handlers.sliding_sync.extensions import SlidingSyncExtensionHandler from synapse.logging.opentracing import log_kv, set_tag, start_active_span, trace from synapse.types import Requester, SlidingSyncStreamToken, StreamToken from synapse.types.handlers.paginated_sync import ( @@ -40,6 +42,7 @@ from synapse.types.handlers.paginated_sync import ( ) from synapse.types.handlers.sliding_sync import ( HaveSentRoomFlag, + PerConnectionState, RoomSyncConfig, SlidingSyncResult, ) @@ -59,6 +62,21 @@ logger = logging.getLogger(__name__) AGING_LANE_FRACTION = 4 +class PaginatedSyncExtensionHandler(SlidingSyncExtensionHandler): + """The sliding sync extensions without the `lists`/`rooms` scoping: with no + lists and no subscriptions there is nothing to scope, so an enabled + extension simply applies to the rooms in the response.""" + + def find_relevant_room_ids_for_extension( + self, + requested_lists: object, + requested_room_ids: object, + actual_lists: object, + actual_room_ids: "set[str] | frozenset[str]", + ) -> set[str]: + return set(actual_room_ids) + + class PaginatedSyncHandler(SlidingSyncHandler): def __init__(self, hs: "HomeServer"): super().__init__(hs) @@ -67,6 +85,13 @@ class PaginatedSyncHandler(SlidingSyncHandler): # historical events because a room's effective limit grew. self.expanded_timeline_on_limit_increase = False + # `required_state` is immutable for the life of a connection, so there + # are no per-room request configs to remember or diff. + self.track_room_configs = False + + # Extensions lose their scoping fields. + self.extensions = PaginatedSyncExtensionHandler(hs) + async def wait_for_paginated_sync_for_user( self, requester: Requester, @@ -160,14 +185,23 @@ class PaginatedSyncHandler(SlidingSyncHandler): limit = sync_config.limit history = sync_config.history if sync_config.history is not None else limit - # Raises SlidingSyncUnknownPosition if the position is unrecognised - # (e.g. the server lost the connection state); the client then starts a - # fresh connection and rooms simply come down as never-sent again. - previous_connection_state = ( - await self.connection_store.get_and_clear_connection_positions( - sync_config, from_token + # There is no M_UNKNOWN_POS in this API: a `pos` the server doesn't + # recognise (expired, forged, another device's) is treated as absent - + # nothing is trusted from the token, the connection starts afresh and + # rooms come down as never-sent. The client has no error path. + try: + previous_connection_state = ( + await self.connection_store.get_and_clear_connection_positions( + sync_config, from_token + ) ) - ) + except SlidingSyncUnknownPosition: + logger.info( + "Unrecognised paginated sync pos for %s; starting the connection afresh", + user_id, + ) + from_token = None + previous_connection_state = PerConnectionState(last_used_ts=None) # Reuse the sliding sync membership machinery wholesale: with no lists # and no subscriptions it assembles the full membership map (with diff --git a/synapse/rest/client/paginated_sync.py b/synapse/rest/client/paginated_sync.py index e770055ebf..16f6636bb1 100644 --- a/synapse/rest/client/paginated_sync.py +++ b/synapse/rest/client/paginated_sync.py @@ -140,6 +140,11 @@ class PaginatedSyncRestServlet(SlidingSyncRestServlet): response["pos"] = await result.next_pos.to_string(self.store) response["rooms"] = await self.encode_rooms(requester, result.rooms) + # `num_live` is derivable in this API (previously-sent rooms only ever + # receive live events; `initial` rooms are all-historical), so it is + # not part of the response. + for room in response["rooms"].values(): + room.pop("num_live", None) response["extensions"] = await self.encode_extensions( requester, result.extensions, result.rooms ) diff --git a/synapse/types/rest/client/__init__.py b/synapse/types/rest/client/__init__.py index aab7b275a0..351d15013b 100644 --- a/synapse/types/rest/client/__init__.py +++ b/synapse/types/rest/client/__init__.py @@ -529,9 +529,14 @@ class PaginatedSyncBody(RequestBodyModel): which has not previously been sent on this connection. Defaults to `limit`. required_state: Required state for each room returned, with the same - semantics as sliding sync's `required_state`. Applied to every room. + semantics as sliding sync's `required_state`. Applied to every + room. Must be identical on every request of a connection; the + server always uses the current request's value and never persists + or diffs it. extensions: Extensions API. A map of extension key to extension config, - shared with sliding sync. + shared with sliding sync. The per-extension `lists`/`rooms` scoping + fields are accepted but ignored: an enabled extension applies to + the rooms in the response. """ conn_id: StrictStr | None = None diff --git a/tests/rest/client/sliding_sync/test_paginated_sync.py b/tests/rest/client/sliding_sync/test_paginated_sync.py index 4f4637e5c1..bc5cdf8b9f 100644 --- a/tests/rest/client/sliding_sync/test_paginated_sync.py +++ b/tests/rest/client/sliding_sync/test_paginated_sync.py @@ -21,7 +21,7 @@ from unittest.mock import AsyncMock from twisted.internet.testing import MemoryReactor import synapse.rest.admin -from synapse.rest.client import login, paginated_sync, room, sync +from synapse.rest.client import login, paginated_sync, receipts, room, sync from synapse.server import HomeServer from synapse.types import JsonDict from synapse.util.clock import Clock @@ -41,6 +41,7 @@ class PaginatedSyncTestCase(unittest.HomeserverTestCase): servlets = [ synapse.rest.admin.register_servlets, login.register_servlets, + receipts.register_servlets, room.register_servlets, sync.register_servlets, paginated_sync.register_servlets, @@ -163,6 +164,8 @@ class PaginatedSyncTestCase(unittest.HomeserverTestCase): self.assertTrue(room_response["limited"]) self.assertIn("prev_batch", room_response) self.assertNotIn("initial", room_response) + # `num_live` is derivable and not part of this API. + self.assertNotIn("num_live", room_response) def test_incremental_backlog_is_paged_and_never_lost(self) -> None: """When more rooms have updates than fit in `page_size`, the rest are @@ -234,3 +237,59 @@ class PaginatedSyncTestCase(unittest.HomeserverTestCase): state_types = {event["type"] for event in room_response["required_state"]} self.assertIn("m.room.create", state_types) self.assertIn("m.room.member", state_types) + + def test_unknown_pos_starts_afresh(self) -> None: + """There is no M_UNKNOWN_POS: a pos the server doesn't recognise is + treated as absent, and rooms come down as never-sent again.""" + room_ids = self._create_rooms(3) + + body = {"page_size": 10, "limit": 5, "history": 1} + response = self._sync(body) + self.assertEqual(len(response["rooms"]), 3) + + # Corrupt the connection position (keep the stream token valid). + connection_position, stream_token = response["pos"].split("/", 1) + bogus_pos = f"{int(connection_position) + 999}/{stream_token}" + + response = self._sync(body, pos=bogus_pos) + + # Not an error: a fresh connection, with every room initial again. + self.assertEqual(set(response["rooms"].keys()), set(room_ids)) + for room_id, room_response in response["rooms"].items(): + self.assertTrue(room_response.get("initial"), room_id) + + def test_extensions_apply_without_scoping(self) -> None: + """Extensions have no lists/rooms scoping: enabling one is enough for + it to apply to the rooms in the response.""" + room_id = self.helper.create_room_as(self.user, tok=self.tok) + + user2 = self.register_user("bob", "password") + tok2 = self.login("bob", "password") + self.helper.join(room_id, user2, tok=tok2) + + event_response = self.helper.send(room_id, body="hello", tok=self.tok) + + body = { + "page_size": 10, + "limit": 5, + "history": 5, + "extensions": {"receipts": {"enabled": True}}, + } + response = self._sync(body) + pos = response["pos"] + + # Bob sends a message and reads the room. + self.helper.send(room_id, body="reply", tok=tok2) + channel = self.make_request( + "POST", + f"/rooms/{room_id}/receipt/m.read/{event_response['event_id']}", + {}, + access_token=tok2, + ) + self.assertEqual(channel.code, 200, channel.json_body) + + response = self._sync(body, pos=pos) + + self.assertIn(room_id, response["rooms"]) + receipts_response = response["extensions"]["receipts"]["rooms"] + self.assertIn(room_id, receipts_response, receipts_response)