Sliding Sync: Handle timeline limit changes (take 2)

changelog.d/17579.misc

@@ -0,0 +1 @@
+Handle changes in `timeline_limit` in experimental sliding sync.

synapse/handlers/sliding_sync.py

@@ -787,7 +787,19 @@ async def current_sync_for_user(
                 # subscription and have updates we need to send (i.e. either because
                 # we haven't sent the room down, or we have but there are missing
                 # updates).
-                for room_id in relevant_room_map:
+                for room_id, room_config in relevant_room_map.items():
+                    prev_room_sync_config = (
+                        previous_connection_state.previous_room_configs.get(room_id)
+                    )
+                    if prev_room_sync_config is not None:
+                        # Always include rooms whose timeline limit has increased.
+                        if (
+                            prev_room_sync_config.timeline_limit
+                            < room_config.timeline_limit
+                        ):
+                            rooms_should_send.add(room_id)
+                            continue
+
                     status = previous_connection_state.rooms.have_sent_room(room_id)
                     if (
                         # The room was never sent down before so the client needs to know
@@ -819,12 +831,17 @@ async def current_sync_for_user(
                     if room_id in rooms_should_send
                 }
 
+        new_connection_state = previous_connection_state.get_mutable()
+
         @trace
         @tag_args
         async def handle_room(room_id: str) -> None:
+            set_tag("room_id", room_id)
+
             room_sync_result = await self.get_room_sync_data(
                 sync_config=sync_config,
                 previous_connection_state=previous_connection_state,
+                new_connection_state=new_connection_state,
                 room_id=room_id,
                 room_sync_config=relevant_rooms_to_send_map[room_id],
                 room_membership_for_user_at_to_token=room_membership_for_user_map[
@@ -842,8 +859,6 @@ async def handle_room(room_id: str) -> None:
             with start_active_span("sliding_sync.generate_room_entries"):
                 await concurrently_execute(handle_room, relevant_rooms_to_send_map, 10)
 
-        new_connection_state = previous_connection_state.get_mutable()
-
         extensions = await self.get_extensions_response(
             sync_config=sync_config,
             actual_lists=lists,
@@ -1955,6 +1970,7 @@ async def get_room_sync_data(
         self,
         sync_config: SlidingSyncConfig,
         previous_connection_state: "PerConnectionState",
+        new_connection_state: "MutablePerConnectionState",
         room_id: str,
         room_sync_config: RoomSyncConfig,
         room_membership_for_user_at_to_token: _RoomMembershipForUser,
@@ -1999,8 +2015,15 @@ async def get_room_sync_data(
         #    connection before
         #
         # Relevant spec issue: https://github.com/matrix-org/matrix-spec/issues/1917
+        #
+        # We also need to check if the timeline limit has increased, if so we ignore
+        # the from bound for the timeline to send down a larger chunk of
+        # history.
+        #
+        # TODO: Also handle changes to `required_state`
         from_bound = None
         initial = True
+        ignore_timeline_bound = False
         if from_token and not room_membership_for_user_at_to_token.newly_joined:
             room_status = previous_connection_state.rooms.have_sent_room(room_id)
             if room_status.status == HaveSentRoomFlag.LIVE:
@@ -2018,7 +2041,39 @@ async def get_room_sync_data(
 
             log_kv({"sliding_sync.room_status": room_status})
 
-        log_kv({"sliding_sync.from_bound": from_bound, "sliding_sync.initial": initial})
+            prev_room_sync_config = previous_connection_state.previous_room_configs.get(
+                room_id
+            )
+            if prev_room_sync_config is not None:
+                # Check if the timeline limit has increased, if so ignore the
+                # timeline bound and record the change.
+                if (
+                    prev_room_sync_config.timeline_limit
+                    < room_sync_config.timeline_limit
+                ):
+                    ignore_timeline_bound = True
+                    new_connection_state.previous_room_configs[room_id] = (
+                        room_sync_config
+                    )
+
+                if (
+                    room_status.status != HaveSentRoomFlag.LIVE
+                    and prev_room_sync_config.timeline_limit
+                    > room_sync_config.timeline_limit
+                ):
+                    new_connection_state.previous_room_configs[room_id] = (
+                        room_sync_config
+                    )
+
+                # TODO: Record changes in required_state.
+
+        log_kv(
+            {
+                "sliding_sync.from_bound": from_bound,
+                "sliding_sync.initial": initial,
+                "sliding_sync.ignore_timeline_bound": ignore_timeline_bound,
+            }
+        )
 
         # Assemble the list of timeline events
         #
@@ -2055,6 +2110,10 @@ async def get_room_sync_data(
                     room_membership_for_user_at_to_token.event_pos.to_room_stream_token()
                 )
 
+            timeline_from_bound = from_bound
+            if ignore_timeline_bound:
+                timeline_from_bound = None
+
             # For initial `/sync` (and other historical scenarios mentioned above), we
             # want to view a historical section of the timeline; to fetch events by
             # `topological_ordering` (best representation of the room DAG as others were
@@ -2080,7 +2139,7 @@ async def get_room_sync_data(
             pagination_method: PaginateFunction = (
                 # Use `topographical_ordering` for historical events
                 paginate_room_events_by_topological_ordering
-                if from_bound is None
+                if timeline_from_bound is None
                 # Use `stream_ordering` for updates
                 else paginate_room_events_by_stream_ordering
             )
@@ -2090,7 +2149,7 @@ async def get_room_sync_data(
                 # (from newer to older events) starting at to_bound.
                 # This ensures we fill the `limit` with the newest events first,
                 from_key=to_bound,
-                to_key=from_bound,
+                to_key=timeline_from_bound,
                 direction=Direction.BACKWARDS,
                 # We add one so we can determine if there are enough events to saturate
                 # the limit or not (see `limited`)
@@ -2448,6 +2507,47 @@ async def get_room_sync_data(
             if new_bump_event_pos.stream > 0:
                 bump_stamp = new_bump_event_pos.stream
 
+        prev_room_sync_config = previous_connection_state.previous_room_configs.get(
+            room_id
+        )
+        if ignore_timeline_bound:
+            # FIXME: We signal the fact that we're sending down more events to
+            # the client by setting `initial=true` *without* sending down all
+            # the state/metadata again, which is what the proxy does. We should
+            # update the protocol to do something less silly.
+            initial = True
+
+            new_connection_state.previous_room_configs[room_id] = RoomSyncConfig(
+                timeline_limit=len(timeline_events),
+                required_state_map=room_sync_config.required_state_map,
+            )
+        elif prev_room_sync_config is not None:
+            # If the result isn't limited then we don't need to record that the
+            # timeline_limit has been reduced, as the *effective* timeline limit
+            # (i.e. the amount of timeline we have previously sent) is at least
+            # the previous timeline limit.
+            #
+            # This is to handle the case where the timeline limit e.g. goes from
+            # 10 to 5 to 10 again (without any timeline gaps), where there's no
+            # point sending down extra events when the timeline limit is
+            # increased as the client already has the 10 previous events.
+            # However, if is a gap (i.e. limited is True), then we *do* need to
+            # record the reduced timeline.
+            if (
+                limited
+                and prev_room_sync_config.timeline_limit
+                > room_sync_config.timeline_limit
+            ):
+                new_connection_state.previous_room_configs[room_id] = RoomSyncConfig(
+                    timeline_limit=len(timeline_events),
+                    required_state_map=room_sync_config.required_state_map,
+                )
+
+            # TODO: Record changes in required_state.
+
+        else:
+            new_connection_state.previous_room_configs[room_id] = room_sync_config
+
         set_tag(SynapseTags.RESULT_PREFIX + "initial", initial)
 
         return SlidingSyncResult.RoomResult(
@@ -3262,16 +3362,32 @@ class PerConnectionState:
     Attributes:
         rooms: The status of each room for the events stream.
         receipts: The status of each room for the receipts stream.
+        previous_room_configs: Map from room_id to the `RoomSyncConfig` of all
+            rooms that we have previously sent down.
     """
 
     rooms: RoomStatusMap[RoomStreamToken] = attr.Factory(RoomStatusMap)
     receipts: RoomStatusMap[MultiWriterStreamToken] = attr.Factory(RoomStatusMap)
 
+    previous_room_configs: Mapping[str, RoomSyncConfig] = attr.Factory(dict)
+
     def get_mutable(self) -> "MutablePerConnectionState":
         """Get a mutable copy of this state."""
+        previous_room_configs = cast(
+            MutableMapping[str, RoomSyncConfig], self.previous_room_configs
+        )
+
         return MutablePerConnectionState(
             rooms=self.rooms.get_mutable(),
             receipts=self.receipts.get_mutable(),
+            previous_room_configs=ChainMap({}, previous_room_configs),
+        )
+
+    def copy(self) -> "PerConnectionState":
+        return PerConnectionState(
+            rooms=self.rooms.copy(),
+            receipts=self.receipts.copy(),
+            previous_room_configs=dict(self.previous_room_configs),
         )
 
 
@@ -3282,8 +3398,18 @@ class MutablePerConnectionState(PerConnectionState):
     rooms: MutableRoomStatusMap[RoomStreamToken]
     receipts: MutableRoomStatusMap[MultiWriterStreamToken]
 
+    previous_room_configs: typing.ChainMap[str, RoomSyncConfig]
+
     def has_updates(self) -> bool:
-        return bool(self.rooms.get_updates()) or bool(self.receipts.get_updates())
+        return (
+            bool(self.rooms.get_updates())
+            or bool(self.receipts.get_updates())
+            or bool(self.get_room_config_updates())
+        )
+
+    def get_room_config_updates(self) -> Mapping[str, RoomSyncConfig]:
+        """Get updates to the room sync config"""
+        return self.previous_room_configs.maps[0]
 
 
 @attr.s(auto_attribs=True)
@@ -3367,7 +3493,6 @@ async def record_new_state(
     ) -> int:
         """Record updated per-connection state, returning the connection
         position associated with the new state.
-
         If there are no changes to the state this may return the same token as
         the existing per-connection state.
         """
@@ -3388,10 +3513,7 @@ async def record_new_state(
 
         # We copy the `MutablePerConnectionState` so that the inner `ChainMap`s
         # don't grow forever.
-        sync_statuses[new_store_token] = PerConnectionState(
-            rooms=new_connection_state.rooms.copy(),
-            receipts=new_connection_state.receipts.copy(),
-        )
+        sync_statuses[new_store_token] = new_connection_state.copy()
 
         return new_store_token
 

tests/rest/client/sliding_sync/test_rooms_timeline.py

@@ -17,6 +17,7 @@
 from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
+from synapse.api.constants import EventTypes
 from synapse.rest.client import login, room, sync
 from synapse.server import HomeServer
 from synapse.types import StreamToken, StrSequence
@@ -573,3 +574,126 @@ def test_rooms_ban_incremental_sync2(self) -> None:
 
         # Nothing to see for this banned user in the room in the token range
         self.assertIsNone(response_body["rooms"].get(room_id1))
+
+    def test_increasing_timeline_range_sends_more_messages(self) -> None:
+        """
+        Test that increasing the timeline limit via room subscriptions sends the
+        room down with more messages in a limited sync.
+        """
+
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        room_id1 = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        sync_body = {
+            "lists": {
+                "foo-list": {
+                    "ranges": [[0, 1]],
+                    "required_state": [[EventTypes.Create, ""]],
+                    "timeline_limit": 1,
+                }
+            }
+        }
+
+        message_events = []
+        for _ in range(10):
+            resp = self.helper.send(room_id1, "msg", tok=user1_tok)
+            message_events.append(resp["event_id"])
+
+        # Make the first Sliding Sync request
+        response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
+        room_response = response_body["rooms"][room_id1]
+
+        self.assertEqual(room_response["initial"], True)
+        self.assertEqual(room_response["limited"], True)
+
+        # We only expect the last message at first
+        self.assertEqual(
+            [event["event_id"] for event in room_response["timeline"]],
+            message_events[-1:],
+            room_response["timeline"],
+        )
+
+        # We also expect to get the create event state.
+        self.assertEqual(
+            [event["type"] for event in room_response["required_state"]],
+            [EventTypes.Create],
+        )
+
+        # Now do another request with a room subscription with an increased timeline limit
+        sync_body["room_subscriptions"] = {
+            room_id1: {
+                "required_state": [],
+                "timeline_limit": 10,
+            }
+        }
+
+        response_body, from_token = self.do_sync(
+            sync_body, since=from_token, tok=user1_tok
+        )
+        room_response = response_body["rooms"][room_id1]
+
+        self.assertEqual(room_response["initial"], True)
+        self.assertEqual(room_response["limited"], True)
+
+        # Now we expect all the messages
+        self.assertEqual(
+            [event["event_id"] for event in room_response["timeline"]],
+            message_events,
+            room_response["timeline"],
+        )
+
+        # We don't expect to get the room create down, as nothing has changed.
+        self.assertNotIn("required_state", room_response)
+
+        # Decreasing the timeline limit shouldn't resend any events
+        sync_body["room_subscriptions"] = {
+            room_id1: {
+                "required_state": [],
+                "timeline_limit": 5,
+            }
+        }
+
+        event_response = self.helper.send(room_id1, "msg", tok=user1_tok)
+        latest_event_id = event_response["event_id"]
+
+        response_body, from_token = self.do_sync(
+            sync_body, since=from_token, tok=user1_tok
+        )
+        room_response = response_body["rooms"][room_id1]
+
+        self.assertNotIn("initial", room_response)
+        self.assertEqual(room_response["limited"], False)
+
+        self.assertEqual(
+            [event["event_id"] for event in room_response["timeline"]],
+            [latest_event_id],
+            room_response["timeline"],
+        )
+
+        # Increasing the limit to what it was before also should not resend any
+        # events
+        sync_body["room_subscriptions"] = {
+            room_id1: {
+                "required_state": [],
+                "timeline_limit": 10,
+            }
+        }
+
+        event_response = self.helper.send(room_id1, "msg", tok=user1_tok)
+        latest_event_id = event_response["event_id"]
+
+        response_body, from_token = self.do_sync(
+            sync_body, since=from_token, tok=user1_tok
+        )
+        room_response = response_body["rooms"][room_id1]
+
+        self.assertNotIn("initial", room_response)
+        self.assertEqual(room_response["limited"], False)
+
+        self.assertEqual(
+            [event["event_id"] for event in room_response["timeline"]],
+            [latest_event_id],
+            room_response["timeline"],
+        )