diff --git a/changelog.d/19734.bugfix b/changelog.d/19734.bugfix new file mode 100644 index 00000000000..01af7d9ab83 --- /dev/null +++ b/changelog.d/19734.bugfix @@ -0,0 +1 @@ +Update Sliding Sync to return a new response immediately if a room subscription have changed and produced a new response. diff --git a/changelog.d/19792.bugfix b/changelog.d/19792.bugfix index 6aba7b21a61..01af7d9ab83 100644 --- a/changelog.d/19792.bugfix +++ b/changelog.d/19792.bugfix @@ -1 +1 @@ -Have SSS return a new response immediately if a room subscription have changed and produced a new response. +Update Sliding Sync to return a new response immediately if a room subscription have changed and produced a new response. diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 98b74b098d9..d10fed2d794 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -213,6 +213,13 @@ async def current_sync_callback( sync_config.user.to_string(), timeout_ms, current_sync_callback, + # We *wait* from `now_token` as we have already computed the sync + # response up to `now_token` above, so as a minor optimization, we + # can wait for something new to arrive after `now_token`. + # + # We still generate the sync response using `from_token` in the + # callback above though, as to generate the correct response it + # needs to know the "real" `from_token`. from_token=now_token, ) did_wait = True diff --git a/tests/rest/client/sliding_sync/test_rooms_required_state.py b/tests/rest/client/sliding_sync/test_rooms_required_state.py index 586b127f8a1..901f22a35d1 100644 --- a/tests/rest/client/sliding_sync/test_rooms_required_state.py +++ b/tests/rest/client/sliding_sync/test_rooms_required_state.py @@ -25,8 +25,10 @@ from synapse.server import HomeServer from synapse.storage.databases.main.events import DeltaState, SlidingSyncTableChanges from synapse.util.clock import Clock +from synapse.util.duration import Duration from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase +from tests.server import TimedOutException from tests.test_utils.event_injection import mark_event_as_partial_state logger = logging.getLogger(__name__) @@ -1924,7 +1926,12 @@ def test_rooms_required_state_expand(self) -> None: def test_rooms_required_state_expand_retract_expand(self) -> None: """Test that when expanding, retracting and then expanding the required - state, we get the changes that happened.""" + state, we get the changes that happened. + + Also see `test_changing_required_state_returns_immediately`, which tests + that the sync stream is woken up immediately when changing the required + state, and not just on the next change to the room. + """ user1_id = self.register_user("user1", "pass") user1_tok = self.login(user1_id, "pass") @@ -2245,3 +2252,75 @@ def test_lazy_loading_room_members_state_reset_non_limited_timeline(self) -> Non response_body["rooms"][room_id]["required_state"][0]["event_id"], first_event_id, ) + + def test_changing_required_state_returns_immediately(self) -> None: + """Test that if we change the `required_state`, then we return immediately + with the new `required_state`.""" + + 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) + + # Make an initial sync request with no required state + sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, + } + } + } + response_body, from_token = self.do_sync(sync_body, tok=user1_tok) + + # We should see no required state + self.assertIsNone(response_body["rooms"][room_id1].get("required_state")) + + # Get the state_map before we change the state as this is the final state we + # expect to see when we update the required state. + state_map = self.get_success( + self.storage_controllers.state.get_current_state(room_id1) + ) + + # There is no new data, and so making another sync request will block. + channel = self.make_sync_request( + sync_body, + since=from_token, + tok=user1_tok, + timeout=Duration(seconds=10), + await_result=False, + ) + + # Request will block for 10 seconds as there no updates. + with self.assertRaises(TimedOutException): + channel.await_result(timeout_ms=9500) + + # Wait for the request to actually finish. (We do this to ensure log + # contexts don't leak between tests). + channel.await_result(timeout_ms=1000) + + # Now update the Sliding Sync requests to include a `required_state` + # event, and make another sync request. + sync_body["lists"]["foo-list"]["required_state"] = [ + [EventTypes.Create, ""], + ] + + channel = self.make_sync_request( + sync_body, + since=from_token, + tok=user1_tok, + timeout=Duration(seconds=10), + await_result=False, + ) + + # We should see the new `required_state` immediately without waiting + channel.await_result(timeout_ms=0) + response_body = channel.json_body + self._assertRequiredStateIncludes( + response_body["rooms"][room_id1]["required_state"], + { + state_map[(EventTypes.Create, "")], + }, + exact=True, + ) diff --git a/tests/rest/client/sliding_sync/test_sliding_sync.py b/tests/rest/client/sliding_sync/test_sliding_sync.py index ebf41cd87c3..fc7d6a279c3 100644 --- a/tests/rest/client/sliding_sync/test_sliding_sync.py +++ b/tests/rest/client/sliding_sync/test_sliding_sync.py @@ -12,6 +12,7 @@ # . # import logging +import urllib.parse from typing import Any, Iterable, Literal from unittest.mock import AsyncMock @@ -43,6 +44,7 @@ StreamToken, ) from synapse.util.clock import Clock +from synapse.util.duration import Duration from synapse.util.stringutils import random_string from tests import unittest @@ -82,7 +84,13 @@ def default_config(self) -> JsonDict: return config def make_sync_request( - self, sync_body: JsonDict, *, since: str | None = None, tok: str + self, + sync_body: JsonDict, + *, + since: str | None = None, + tok: str, + timeout: Duration | None = None, + await_result: bool = True, ) -> FakeChannel: """Make a sliding sync request with given body. @@ -90,25 +98,40 @@ def make_sync_request( sync_body: The full request body to use since: Optional since token tok: Access token to use - + timeout_ms: Optional timeout in milliseconds to use for the request. + await_result: Whether to block and wait for the result before returning. Returns: A tuple of the response body and the `pos` field. """ sync_path = self.sync_endpoint + + query_params: dict[str, Any] = {} if since: - sync_path += f"?pos={since}" + query_params["pos"] = since + if timeout is not None: + query_params["timeout"] = timeout.as_millis() + + if query_params: + query_str = urllib.parse.urlencode(query_params) + sync_path += f"?{query_str}" channel = self.make_request( method="POST", path=sync_path, content=sync_body, access_token=tok, + await_result=await_result, ) return channel def do_sync( - self, sync_body: JsonDict, *, since: str | None = None, tok: str + self, + sync_body: JsonDict, + *, + since: str | None = None, + tok: str, + timeout: Duration | None = None, ) -> tuple[JsonDict, str]: """Do a sliding sync request with given body. @@ -118,11 +141,14 @@ def do_sync( sync_body: The full request body to use since: Optional since token tok: Access token to use + timeout: Optional timeout to use for the request. Returns: A tuple of the response body and the `pos` field. """ - channel = self.make_sync_request(sync_body, since=since, tok=tok) + channel = self.make_sync_request( + sync_body, since=since, tok=tok, timeout=timeout + ) self.assertEqual(channel.code, 200, channel.json_body) return channel.json_body, channel.json_body["pos"]