Reintroduce #19714 - Send a SSS response immediately if the config has changed

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

@@ -0,0 +1 @@
+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

@@ -184,34 +184,45 @@ async def wait_for_sync_for_user(
                 timeout_ms -= after_wait_ts - before_wait_ts
                 timeout_ms = max(timeout_ms, 0)
 
-        # We're going to respond immediately if the timeout is 0 or if this is an
-        # initial sync (without a `from_token`) so we can avoid calling
-        # `notifier.wait_for_events()`.
-        if timeout_ms == 0 or from_token is None:
-            now_token = self.event_sources.get_current_token()
-            result = await self.current_sync_for_user(
+        # Compute a response immediately. We always need to do this before
+        # waiting for new data (unlike in /v3/sync), as the request config might
+        # have changed (e.g. new room subscriptions, etc).
+        now_token = self.event_sources.get_current_token()
+        result = await self.current_sync_for_user(
+            sync_config,
+            from_token=from_token,
+            to_token=now_token,
+        )
+
+        # Return immediately if we have a result, the timeout is 0, or this is
+        # an initial sync.
+        if result or timeout_ms == 0 or from_token is None:
+            return result, did_wait
+
+        # Otherwise, we wait for something to happen and report it to the user.
+        async def current_sync_callback(
+            before_token: StreamToken, after_token: StreamToken
+        ) -> SlidingSyncResult:
+            return await self.current_sync_for_user(
                 sync_config,
                 from_token=from_token,
-                to_token=now_token,
+                to_token=after_token,
             )
-        else:
-            # Otherwise, we wait for something to happen and report it to the user.
-            async def current_sync_callback(
-                before_token: StreamToken, after_token: StreamToken
-            ) -> SlidingSyncResult:
-                return await self.current_sync_for_user(
-                    sync_config,
-                    from_token=from_token,
-                    to_token=after_token,
-                )
 
-            result = await self.notifier.wait_for_events(
-                sync_config.user.to_string(),
-                timeout_ms,
-                current_sync_callback,
-                from_token=from_token.stream_token,
-            )
-            did_wait = True
+        result = await self.notifier.wait_for_events(
+            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
 
         return result, did_wait
 

synapse/handlers/sliding_sync/room_lists.py

@@ -852,11 +852,15 @@ async def _filter_relevant_rooms_to_send(
                             previous_connection_state.room_configs.get(room_id)
                         )
                         if prev_room_sync_config is not None:
-                            # Always include rooms whose timeline limit has increased.
-                            # (see the "XXX: Odd behavior" described below)
+                            # Always include rooms whose effective config has
+                            # expanded. This covers timeline-limit increases and
+                            # required-state additions introduced by room
+                            # subscriptions overriding list-derived params.
                             if (
-                                prev_room_sync_config.timeline_limit
-                                < room_config.timeline_limit
+                                prev_room_sync_config.combine_room_sync_config(
+                                    room_config
+                                )
+                                != prev_room_sync_config
                             ):
                                 rooms_should_send.add(room_id)
                                 continue

synapse/types/handlers/sliding_sync.py

@@ -203,6 +203,9 @@ class StrippedHero:
         highlight_count: int
 
         def __bool__(self) -> bool:
+            """Are there any updates that should be returned immediately to
+            the client?
+            """
             return (
                 # If this is the first time the client is seeing the room, we should not filter it out
                 # under any circumstance.
@@ -270,6 +273,8 @@ class ToDeviceExtension:
             events: Sequence[JsonMapping]
 
             def __bool__(self) -> bool:
+                """Are there any updates that should be returned immediately to
+                the client?"""
                 return bool(self.events)
 
         @attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -294,23 +299,37 @@ class E2eeExtension:
             device_unused_fallback_key_types: Sequence[str]
 
             def __bool__(self) -> bool:
-                # Note that "signed_curve25519" is always returned in key count responses
-                # regardless of whether we uploaded any keys for it. This is necessary until
+                """Are there any updates that should be returned immediately to
+                the client?"""
+                # Note that "signed_curve25519" is always returned in key count
+                # responses regardless of whether we uploaded any keys for it.
+                # This is necessary until
                 # https://github.com/matrix-org/matrix-doc/issues/3298 is fixed.
                 #
                 # Also related:
                 # https://github.com/element-hq/element-android/issues/3725 and
                 # https://github.com/matrix-org/synapse/issues/10456
-                default_otk = self.device_one_time_keys_count.get("signed_curve25519")
-                more_than_default_otk = len(self.device_one_time_keys_count) > 1 or (
-                    default_otk is not None and default_otk > 0
-                )
-
-                return bool(
-                    more_than_default_otk
-                    or self.device_list_updates
-                    or self.device_unused_fallback_key_types
-                )
+                #
+                # This is why we don't incorporate `device_one_time_keys_count`
+                # (or `device_unused_fallback_key_types`) into the `__bool__`
+                # check.
+                #
+                # FIXME: Ideally we'd detect if either of those fields have
+                # changed since the last sync, but we do not currently track
+                # such state.
+                #
+                # Note that the client will receive these fields eventually when
+                # we respond to the sync request (usually sync timeouts are set
+                # to ~30s), we just won't immediately respond (even if there are
+                # changes). This delay is acceptable for clients, as a) these
+                # fields do not trigger UI (and so don't affect user perceivable
+                # latency) and b) are handled in the background by the clients
+                # anyway. The only risk being that one-time keys could be exhausted
+                # before the client knows about adding some more. But for example,
+                # if the client is syncing with a timeout of 30s, the window of
+                # staleness is so small for this not to matter.
+
+                return bool(self.device_list_updates)
 
         @attr.s(slots=True, frozen=True, auto_attribs=True)
         class AccountDataExtension:
@@ -327,6 +346,8 @@ class AccountDataExtension:
             account_data_by_room_map: Mapping[str, Mapping[str, JsonMapping]]
 
             def __bool__(self) -> bool:
+                """Are there any updates that should be returned immediately to
+                the client?"""
                 return bool(
                     self.global_account_data_map or self.account_data_by_room_map
                 )
@@ -343,6 +364,8 @@ class ReceiptsExtension:
             room_id_to_receipt_map: Mapping[str, JsonMapping]
 
             def __bool__(self) -> bool:
+                """Are there any updates that should be returned immediately to
+                the client?"""
                 return bool(self.room_id_to_receipt_map)
 
         @attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -357,6 +380,8 @@ class TypingExtension:
             room_id_to_typing_map: Mapping[str, JsonMapping]
 
             def __bool__(self) -> bool:
+                """Are there any updates that should be returned immediately to
+                the client?"""
                 return bool(self.room_id_to_typing_map)
 
         @attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -391,6 +416,8 @@ class ThreadUnsubscription:
             prev_batch: ThreadSubscriptionsToken | None
 
             def __bool__(self) -> bool:
+                """Are there any updates that should be returned immediately to
+                the client?"""
                 return (
                     bool(self.subscribed)
                     or bool(self.unsubscribed)
@@ -405,6 +432,8 @@ def __bool__(self) -> bool:
         thread_subscriptions: ThreadSubscriptionsExtension | None = None
 
         def __bool__(self) -> bool:
+            """Are there any updates that should be returned immediately to
+            the client?"""
             return bool(
                 self.to_device
                 or self.e2ee
@@ -420,9 +449,14 @@ def __bool__(self) -> bool:
     extensions: Extensions
 
     def __bool__(self) -> bool:
-        """Make the result appear empty if there are no updates. This is used
-        to tell if the notifier needs to wait for more events when polling for
-        events.
+        """Are there any updates that should be returned immediately to
+        the client?
+
+        This is used to determine if a sliding sync response should be returned
+        immediately or if the notifier needs to wait for further updates, and
+        thus MUST return false if there is no new data since the last sync. This
+        is subtly different than just checking if any of the fields are set,
+        since some fields are always included (like `bump_stamp`).
         """
         # We don't include `self.lists` here, as a) `lists` is always non-empty even if
         # there are no changes, and b) since we're sorting rooms by `stream_ordering` of

tests/rest/client/sliding_sync/test_extension_e2ee.py

@@ -290,6 +290,173 @@ def test_wait_for_new_data_timeout(self) -> None:
             [],
         )
 
+    def test_wait_for_new_data_timeout_with_otks(self) -> None:
+        """
+        Test that an incremental Sliding Sync with the e2ee extension enabled
+        does not return immediately when the user has uploaded one-time keys
+        (i.e. `device_one_time_keys_count` contains entries beyond the default
+        `signed_curve25519: 0`).
+        """
+        test_device_id = "TESTDEVICE"
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass", device_id=test_device_id)
+
+        # Upload one-time keys for the user/device so that
+        # `device_one_time_keys_count` is non-default in the response.
+        self.get_success(
+            self.e2e_keys_handler.upload_keys_for_user(
+                user1_id,
+                test_device_id,
+                {
+                    "one_time_keys": {
+                        "alg1:k1": "key1",
+                        "alg2:k2": {"key": "key2", "signatures": {"k1": "sig1"}},
+                    }
+                },
+            )
+        )
+
+        sync_body = {
+            "lists": {},
+            "extensions": {
+                "e2ee": {
+                    "enabled": True,
+                }
+            },
+        }
+        _, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        # Make an incremental Sliding Sync request with a timeout
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint + f"?timeout=10000&pos={from_token}",
+            content=sync_body,
+            access_token=user1_tok,
+            await_result=False,
+        )
+        # Block for 5 seconds to make sure we are `notifier.wait_for_events(...)`
+        with self.assertRaises(TimedOutException):
+            channel.await_result(timeout_ms=5000)
+
+        # Wake-up `notifier.wait_for_events(...)` that will cause us to test
+        # `SlidingSyncResult.__bool__` for new results. The non-default
+        # `device_one_time_keys_count` must not be considered new data.
+        self._bump_notifier_wait_for_events(
+            user1_id, wake_stream_key=StreamKeyType.ACCOUNT_DATA
+        )
+
+        # Block for a little bit more to ensure we don't see any new results.
+        with self.assertRaises(TimedOutException):
+            channel.await_result(timeout_ms=4000)
+        # Wait for the sync to complete (wait for the rest of the 10 second timeout,
+        # 5000 + 4000 + 1200 > 10000)
+        channel.await_result(timeout_ms=1200)
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        # Device lists are present for incremental syncs but empty because no device changes
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"]
+            .get("device_lists", {})
+            .get("changed"),
+            [],
+        )
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"].get("device_lists", {}).get("left"),
+            [],
+        )
+
+        # The one-time key counts are present, but they should not have caused
+        # the sync to return early.
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"]["device_one_time_keys_count"],
+            {
+                "alg1": 1,
+                "alg2": 1,
+                "signed_curve25519": 0,
+            },
+        )
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"]["device_unused_fallback_key_types"],
+            [],
+        )
+
+    def test_wait_for_new_data_timeout_with_fallback_keys(self) -> None:
+        """
+        Test that an incremental Sliding Sync with the e2ee extension enabled
+        does not return immediately when the user has uploaded fallback keys
+        (i.e. `device_unused_fallback_key_types` is non-empty).
+        """
+        test_device_id = "TESTDEVICE"
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass", device_id=test_device_id)
+
+        # Upload a fallback key for the user/device so that
+        # `device_unused_fallback_key_types` is non-empty in the response.
+        self.get_success(
+            self.e2e_keys_handler.upload_keys_for_user(
+                user1_id,
+                test_device_id,
+                {"fallback_keys": {"alg1:k1": "fallback_key1"}},
+            )
+        )
+
+        sync_body = {
+            "lists": {},
+            "extensions": {
+                "e2ee": {
+                    "enabled": True,
+                }
+            },
+        }
+        _, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        # Make an incremental Sliding Sync request with a timeout
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint + f"?timeout=10000&pos={from_token}",
+            content=sync_body,
+            access_token=user1_tok,
+            await_result=False,
+        )
+        # Block for 5 seconds to make sure we are `notifier.wait_for_events(...)`
+        with self.assertRaises(TimedOutException):
+            channel.await_result(timeout_ms=5000)
+
+        # Wake-up `notifier.wait_for_events(...)` that will cause us to test
+        # `SlidingSyncResult.__bool__` for new results. The non-empty
+        # `device_unused_fallback_key_types` must not be considered new data.
+        self._bump_notifier_wait_for_events(
+            user1_id, wake_stream_key=StreamKeyType.ACCOUNT_DATA
+        )
+
+        # Block for a little bit more to ensure we don't see any new results.
+        with self.assertRaises(TimedOutException):
+            channel.await_result(timeout_ms=4000)
+
+        # Wait for the sync to complete (wait for the rest of the 10 second timeout,
+        # 5000 + 4000 + 1200 > 10000)
+        channel.await_result(timeout_ms=1200)
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        # Device lists are present for incremental syncs but empty because no device changes
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"]
+            .get("device_lists", {})
+            .get("changed"),
+            [],
+        )
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"].get("device_lists", {}).get("left"),
+            [],
+        )
+
+        # The unused fallback key types are present, but they should not have
+        # caused the sync to return early.
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"]["device_unused_fallback_key_types"],
+            ["alg1"],
+        )
+
     def test_device_lists(self) -> None:
         """
         Test that device list updates are included in the response