From 20feb1ef15c6a3370d2b5b276b61819def79c037 Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Fri, 7 Aug 2026 13:38:08 +0300 Subject: [PATCH] Apply PR review: advertise org.matrix.msc4525 in /versions, tolerate malformed pos, protect cold-start drain, fix total_rooms - /versions now advertises org.matrix.msc4525, honouring both the global flag and the per-user experimental feature (as Erik requested inline). - An unparsable pos no longer 400s: it is treated like an unrecognised one and the connection starts afresh (MSC4525 has no client error path). - The aging lane now also reserves page slots for never-sent rooms, so continuous traffic in already-delivered rooms can't stall the initial drain indefinitely. - total_rooms is counted before partial-state filtering so cold-start progress reporting isn't understated. --- rust/src/config/mod.rs | 1 + rust/src/handlers/versions.rs | 22 +++++++++ rust/src/storage/store.rs | 2 + synapse/handlers/sliding_sync/paginated.py | 48 ++++++++++++++----- synapse/rest/client/paginated_sync.py | 16 +++++-- .../test_msc4525_paginated_sync.py | 29 +++++++++++ tests/rest/client/test_versions.py | 33 +++++++++++++ 7 files changed, 137 insertions(+), 14 deletions(-) diff --git a/rust/src/config/mod.rs b/rust/src/config/mod.rs index d79d12a83a..4b2c5f4b3f 100644 --- a/rust/src/config/mod.rs +++ b/rust/src/config/mod.rs @@ -64,6 +64,7 @@ pub struct ExperimentalConfig { pub msc4108_enabled: bool, pub msc4108_delegation_endpoint: Option, pub msc3575_enabled: bool, + pub msc4525_enabled: bool, pub msc4133_enabled: bool, pub msc4155_enabled: bool, pub msc4306_enabled: bool, diff --git a/rust/src/handlers/versions.rs b/rust/src/handlers/versions.rs index 25da9d23fc..4784bbefb3 100644 --- a/rust/src/handlers/versions.rs +++ b/rust/src/handlers/versions.rs @@ -120,9 +120,27 @@ async fn build_versions_response( None => global_unstable_feature_map.msc3575, }; + let msc4525_enabled = match user_id { + Some(user_id) => { + // Don't both looking anything up if it's enabled for everyone + if global_unstable_feature_map.msc4525 { + true + } else { + // Look up whether it's explicitly enabled/disabled for this user + store + .is_feature_enabled_for_user(user_id, PerUserExperimentalFeature::MSC4525) + .await? + // Default to false if there is no entry for this user + .unwrap_or(false) + } + } + None => global_unstable_feature_map.msc4525, + }; + let unstable_feature_map = UnstableFeatureMap { msc3575: msc3575_enabled, msc3881: msc3881_enabled, + msc4525: msc4525_enabled, ..*global_unstable_feature_map }; @@ -236,6 +254,9 @@ pub struct UnstableFeatureMap { /// Simplified sliding sync #[serde(rename = "org.matrix.simplified_msc3575")] msc3575: bool, + /// MSC4525: Paginated sync + #[serde(rename = "org.matrix.msc4525")] + msc4525: bool, /// Arbitrary key-value profile fields. #[serde(rename = "uk.tcpip.msc4133")] msc4133: bool, @@ -309,6 +330,7 @@ pub fn synapse_config_to_global_unstable_feature_map( || (config.experimental.msc4108_delegation_endpoint.is_some()), msc4140: config.server.msc4140_enabled, msc3575: config.experimental.msc3575_enabled, + msc4525: config.experimental.msc4525_enabled, msc4133: config.experimental.msc4133_enabled, msc4133_stable: true, msc4155: config.experimental.msc4155_enabled, diff --git a/rust/src/storage/store.rs b/rust/src/storage/store.rs index b339d7748c..32deff7da1 100644 --- a/rust/src/storage/store.rs +++ b/rust/src/storage/store.rs @@ -29,6 +29,8 @@ pub enum PerUserExperimentalFeature { MSC3575, #[serde(rename = "msc4222")] MSC4222, + #[serde(rename = "msc4525")] + MSC4525, } impl std::fmt::Display for PerUserExperimentalFeature { diff --git a/synapse/handlers/sliding_sync/paginated.py b/synapse/handlers/sliding_sync/paginated.py index 7bddfaf1c4..31d4500e38 100644 --- a/synapse/handlers/sliding_sync/paginated.py +++ b/synapse/handlers/sliding_sync/paginated.py @@ -230,6 +230,9 @@ class MSC4525PaginatedSyncHandler(SlidingSyncHandler): ) sync_room_map = dict(interested_rooms.room_membership_for_user_map) + # For `total_rooms` (cold-start progress): the account's room count, + # before any partial-state filtering below. + total_rooms = len(sync_room_map) newly_joined_rooms = interested_rooms.newly_joined_rooms newly_left_rooms = interested_rooms.newly_left_rooms dm_room_ids = interested_rooms.dm_room_ids @@ -350,23 +353,46 @@ class MSC4525PaginatedSyncHandler(SlidingSyncHandler): # Page: most recently active rooms first. When the page overflows, a # slice of it is reserved for the longest-deferred rooms so that - # nothing is starved by busier rooms perpetually sorting first. + # nothing - neither a deferred update nor the never-sent cold-start + # backlog - is starved by busier rooms perpetually sorting first. page_room_ids: list[str] = [] if candidates: aged_room_ids: list[str] = [] - if len(candidates) > page_size and previously_room_ids: + if ( + from_token is not None + and len(candidates) > page_size + and (previously_room_ids or never_room_ids) + ): aging_lane_size = max(1, page_size // AGING_LANE_FRACTION) - def last_sent_stream_pos(room_id: str) -> int: - room_status = previous_connection_state.rooms.have_sent_room( - room_id + # Never-sent rooms first (deferred since the connection + # started, and invisible to the client until delivered), most + # recently active first to match the page's ordering; then the + # longest-deferred previously-sent rooms. + if never_room_ids: + never_room_infos = await self.room_lists.sort_rooms( + {room_id: sync_room_map[room_id] for room_id in never_room_ids}, + to_token, + limit=aging_lane_size, + ) + aged_room_ids.extend( + room_info.room_id for room_info in never_room_infos ) - assert room_status.last_token is not None - return room_status.last_token.stream - aged_room_ids = sorted(previously_room_ids, key=last_sent_stream_pos)[ - :aging_lane_size - ] + if len(aged_room_ids) < aging_lane_size and previously_room_ids: + + def last_sent_stream_pos(room_id: str) -> int: + room_status = previous_connection_state.rooms.have_sent_room( + room_id + ) + assert room_status.last_token is not None + return room_status.last_token.stream + + aged_room_ids.extend( + sorted(previously_room_ids, key=last_sent_stream_pos)[ + : aging_lane_size - len(aged_room_ids) + ] + ) sorted_room_infos = await self.room_lists.sort_rooms( {room_id: sync_room_map[room_id] for room_id in candidates}, @@ -466,7 +492,7 @@ class MSC4525PaginatedSyncHandler(SlidingSyncHandler): rooms=rooms, extensions=extensions, pending=pending, - total_rooms=len(sync_room_map), + total_rooms=total_rooms, ) set_tag("paginated_sync.result", bool(result)) diff --git a/synapse/rest/client/paginated_sync.py b/synapse/rest/client/paginated_sync.py index 6ce02a8693..4bf54ce8ca 100644 --- a/synapse/rest/client/paginated_sync.py +++ b/synapse/rest/client/paginated_sync.py @@ -23,6 +23,7 @@ sync servlet; only the request parsing and the top-level response differ. import logging from typing import TYPE_CHECKING +from synapse.api.errors import SynapseError from synapse.http.server import HttpServer from synapse.http.servlet import ( parse_and_validate_json_object_from_request, @@ -80,9 +81,18 @@ class MSC4525PaginatedSyncRestServlet(SlidingSyncRestServlet): from_token = None if from_token_string is not None: - from_token = await SlidingSyncStreamToken.from_string( - self.store, from_token_string - ) + try: + from_token = await SlidingSyncStreamToken.from_string( + self.store, from_token_string + ) + except SynapseError: + # There is no client error path in this API: an unparsable + # `pos` is treated the same as an unrecognised one - as + # absent, so the connection starts afresh. + logger.info( + "Unparsable paginated sync pos for %s; starting the connection afresh", + user, + ) body = parse_and_validate_json_object_from_request( request, MSC4525PaginatedSyncBody diff --git a/tests/rest/client/sliding_sync/test_msc4525_paginated_sync.py b/tests/rest/client/sliding_sync/test_msc4525_paginated_sync.py index d4161306f5..2d00a336ea 100644 --- a/tests/rest/client/sliding_sync/test_msc4525_paginated_sync.py +++ b/tests/rest/client/sliding_sync/test_msc4525_paginated_sync.py @@ -339,6 +339,35 @@ class MSC4525PaginatedSyncTestCase(unittest.HomeserverTestCase): for room_id, room_response in response["rooms"].items(): self.assertTrue(room_response.get("initial"), room_id) + # An entirely unparsable pos is treated the same way, not a 400. + response = self._sync(body, pos="not a token at all") + self.assertEqual(set(response["rooms"].keys()), set(room_ids)) + + def test_cold_start_backlog_not_starved_by_live_traffic(self) -> None: + """Rooms never sent on the connection get a reserved slice of every + page, so continuous traffic in already-sent rooms can't stall the + initial drain indefinitely.""" + room_ids = self._create_rooms(6) + + body = {"page_size": 2, "limit": 5, "history": 1} + response = self._sync(body) + seen_rooms = dict(response["rooms"]) + pos = response["pos"] + self.assertEqual(len(seen_rooms), 2) + + # Keep the already-delivered rooms busy while draining: without the + # aging lane every page would fill with the busy (most recently + # active) rooms and the never-sent backlog would starve. + for _ in range(20): + if not response.get("pending"): + break + for room_id in seen_rooms: + self.helper.send(room_id, body="chatter", tok=self.tok) + response = self._sync(body, pos=pos) + seen_rooms.update(response["rooms"]) + pos = response["pos"] + self.assertEqual(set(seen_rooms.keys()), set(room_ids)) + 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.""" diff --git a/tests/rest/client/test_versions.py b/tests/rest/client/test_versions.py index bbdbe38e07..bb09f6aa17 100644 --- a/tests/rest/client/test_versions.py +++ b/tests/rest/client/test_versions.py @@ -142,6 +142,39 @@ class VersionsTestCase(unittest.HomeserverTestCase): channel.json_body, ) + def test_msc4525_advertised_per_user(self) -> None: + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + + # Disabled by default. + channel = self.make_request( + "GET", + "/_matrix/client/versions", + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) + self.assertEqual( + channel.json_body["unstable_features"]["org.matrix.msc4525"], + False, + channel.json_body, + ) + + # Advertised once enabled for the user. + self._enable_experimental_feature_for_user( + target_user_id=user1_id, features={"msc4525": True} + ) + channel = self.make_request( + "GET", + "/_matrix/client/versions", + access_token=user1_tok, + ) + self.assertEqual(channel.code, 200, channel.result) + self.assertEqual( + channel.json_body["unstable_features"]["org.matrix.msc4525"], + True, + channel.json_body, + ) + def test_msc4446_false_by_default(self) -> None: channel = self.make_request("GET", "/_matrix/client/versions") self.assertEqual(channel.code, 200, channel.result)