From be540f9c0571dbdafac1a08058285c9c178eb625 Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Thu, 6 Aug 2026 22:34:32 +0300 Subject: [PATCH] Add tests for the Paginated Sync endpoint Covers initial-sync paging + drain (pending/total_rooms, most-recent first, no duplication or loss), per-room gapping (limited + prev_batch, only the newest limit events), incremental backlog paging, history semantics for never-sent rooms, and top-level required_state. --- .../sliding_sync/test_paginated_sync.py | 236 ++++++++++++++++++ 1 file changed, 236 insertions(+) create mode 100644 tests/rest/client/sliding_sync/test_paginated_sync.py diff --git a/tests/rest/client/sliding_sync/test_paginated_sync.py b/tests/rest/client/sliding_sync/test_paginated_sync.py new file mode 100644 index 0000000000..4f4637e5c1 --- /dev/null +++ b/tests/rest/client/sliding_sync/test_paginated_sync.py @@ -0,0 +1,236 @@ +# +# 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: +# . +# + +"""Tests for the Paginated Sync endpoint (MSC TBD).""" + +import logging +import urllib.parse +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.server import HomeServer +from synapse.types import JsonDict +from synapse.util.clock import Clock + +from tests import unittest + +logger = logging.getLogger(__name__) + + +class PaginatedSyncTestCase(unittest.HomeserverTestCase): + """ + Tests for `POST /_matrix/client/unstable/org.matrix.paginated_sync/sync`: + paging on initial sync, per-room gapping on incremental sync, backlog + (`pending`) draining, and most-recent-first ordering. + """ + + servlets = [ + synapse.rest.admin.register_servlets, + login.register_servlets, + room.register_servlets, + sync.register_servlets, + paginated_sync.register_servlets, + ] + + sync_endpoint = "/_matrix/client/unstable/org.matrix.paginated_sync/sync" + + def default_config(self) -> JsonDict: + config = super().default_config() + config["experimental_features"] = {"paginated_sync_enabled": True} + return config + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + # Use the new sliding sync tables (c.f. SlidingSyncBase). + hs.get_datastores().main.have_finished_sliding_sync_background_jobs = AsyncMock( # type: ignore[method-assign] + return_value=True + ) + + self.user = self.register_user("alice", "password") + self.tok = self.login("alice", "password") + + def _sync( + self, + body: JsonDict, + *, + pos: str | None = None, + ) -> JsonDict: + path = self.sync_endpoint + query: dict[str, str] = {"timeout": "0"} + if pos is not None: + query["pos"] = pos + path += "?" + urllib.parse.urlencode(query) + + channel = self.make_request( + method="POST", path=path, content=body, access_token=self.tok + ) + self.assertEqual(channel.code, 200, channel.json_body) + return channel.json_body + + def _create_rooms(self, count: int) -> list[str]: + room_ids = [] + for i in range(count): + room_id = self.helper.create_room_as(self.user, tok=self.tok) + self.helper.send(room_id, body=f"message in room {i}", tok=self.tok) + room_ids.append(room_id) + return room_ids + + def _drain(self, body: JsonDict, pos: str) -> tuple[dict[str, JsonDict], str]: + """Keep syncing until `pending` is 0; returns all rooms seen and the + final pos.""" + all_rooms: dict[str, JsonDict] = {} + for _ in range(50): + response = self._sync(body, pos=pos) + all_rooms.update(response["rooms"]) + pos = response["pos"] + if not response.get("pending"): + break + else: + self.fail("Backlog never drained") + return all_rooms, pos + + def test_initial_sync_pages_through_all_rooms(self) -> None: + """An initial sync returns at most `page_size` rooms (most recently + active first), reports the backlog in `pending`, and subsequent + requests drain it without duplication or loss.""" + room_ids = self._create_rooms(25) + + body = {"page_size": 10, "limit": 5, "history": 1} + response = self._sync(body) + + self.assertEqual(len(response["rooms"]), 10, response["rooms"].keys()) + self.assertEqual(response["pending"], 15) + self.assertEqual(response["total_rooms"], 25) + + # Most recently active first: the last 10 rooms created. + self.assertEqual(set(response["rooms"].keys()), set(room_ids[-10:])) + + for room_id, room_response in response["rooms"].items(): + self.assertTrue(room_response["initial"], room_id) + # `history: 1`: one timeline event per room, with a prev_batch to + # fetch more via /messages. + self.assertEqual(len(room_response["timeline"]), 1, room_id) + self.assertIn("prev_batch", room_response) + + # Drain the rest. + seen_rooms, pos = self._drain(body, response["pos"]) + seen_rooms.update(response["rooms"]) + + self.assertEqual(set(seen_rooms.keys()), set(room_ids)) + + # Fully caught up: an immediate re-sync returns nothing. + response = self._sync(body, pos=pos) + self.assertEqual(response["rooms"], {}) + self.assertNotIn("pending", response) + + def test_incremental_sync_gaps_busy_rooms(self) -> None: + """A room with more than `limit` new events comes down with the most + recent `limit` of them, `limited: true` and a `prev_batch` - the gap is + explicit and local to the room.""" + room_ids = self._create_rooms(3) + + body = {"page_size": 10, "limit": 5, "history": 1} + response = self._sync(body) + pos = response["pos"] + + # 8 new events in one room; only the last 5 should come down. + event_ids = [] + for i in range(8): + sent = self.helper.send(room_ids[0], body=f"burst {i}", tok=self.tok) + event_ids.append(sent["event_id"]) + + response = self._sync(body, pos=pos) + + self.assertEqual(set(response["rooms"].keys()), {room_ids[0]}) + room_response = response["rooms"][room_ids[0]] + self.assertEqual( + [event["event_id"] for event in room_response["timeline"]], + event_ids[-5:], + ) + self.assertTrue(room_response["limited"]) + self.assertIn("prev_batch", room_response) + self.assertNotIn("initial", 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 + reported in `pending` and delivered (with their events) on subsequent + requests.""" + room_ids = self._create_rooms(6) + + body = {"page_size": 10, "limit": 5, "history": 1} + response = self._sync(body) + pos = response["pos"] + + # One new event in every room. + expected_event_ids = {} + for i, room_id in enumerate(room_ids): + sent = self.helper.send(room_id, body=f"update {i}", tok=self.tok) + expected_event_ids[room_id] = sent["event_id"] + + # Page through them two at a time. + small_page_body = {"page_size": 2, "limit": 5, "history": 1} + response = self._sync(small_page_body, pos=pos) + self.assertEqual(len(response["rooms"]), 2) + self.assertEqual(response["pending"], 4) + + seen_rooms, _ = self._drain(small_page_body, response["pos"]) + seen_rooms.update(response["rooms"]) + + self.assertEqual(set(seen_rooms.keys()), set(room_ids)) + for room_id, room_response in seen_rooms.items(): + self.assertEqual( + room_response["timeline"][-1]["event_id"], + expected_event_ids[room_id], + room_id, + ) + + def test_newly_joined_room_uses_history(self) -> None: + """A room that appears mid-session (never sent on the connection) comes + down `initial` with `history` events.""" + self._create_rooms(2) + + body = {"page_size": 10, "limit": 5, "history": 2} + response = self._sync(body) + pos = response["pos"] + + new_room_id = self.helper.create_room_as(self.user, tok=self.tok) + for i in range(4): + self.helper.send(new_room_id, body=f"new room message {i}", tok=self.tok) + + response = self._sync(body, pos=pos) + + self.assertIn(new_room_id, response["rooms"]) + room_response = response["rooms"][new_room_id] + self.assertTrue(room_response.get("initial"), room_response) + self.assertEqual(len(room_response["timeline"]), 2) + self.assertIn("prev_batch", room_response) + + def test_required_state_is_returned(self) -> None: + """The top-level `required_state` is applied to every room.""" + self._create_rooms(1) + + body = { + "page_size": 10, + "limit": 5, + "history": 1, + "required_state": [["m.room.create", ""], ["m.room.member", "$ME"]], + } + response = self._sync(body) + + (room_response,) = response["rooms"].values() + 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)