Add clarification and test to fixed SSS behaviour

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.

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.

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

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,
+        )

tests/rest/client/sliding_sync/test_sliding_sync.py

@@ -12,6 +12,7 @@
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 #
 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,33 +84,54 @@ 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.
 
         Attributes:
             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"]