Paginated sync: fold in further simplifications

- No M_UNKNOWN_POS: an unrecognised pos is treated as absent (nothing
  trusted from the token); the connection starts afresh and rooms come
  down as never-sent. Clients have no error path.
- required_state is immutable per connection and always taken from the
  current request: room configs are no longer persisted or diffed
  (track_room_configs flag, off for paginated sync).
- num_live dropped from the wire (derivable: previously-sent rooms only
  receive live events, initial rooms are all-historical).
- Extension lists/rooms scoping ignored: an enabled extension applies
  to the rooms in the response (PaginatedSyncExtensionHandler).

Tests: unknown-pos-starts-afresh, extensions-without-scoping, num_live
absence; sliding sync suites unaffected (174 still green).

Also gitignore localtest/, the local validation server's scratch dir.

.gitignore

@@ -81,3 +81,4 @@ book/
 
 # Don't include users' poetry configs
 /poetry.toml
+/localtest/

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,

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

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
         )

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

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)