Merge branch 'erikj/ss_incr_sync' into erikj/ss_hacks

3 files changed, 209 insertions, 4 deletions

diff --git a/pyproject.toml b/pyproject.toml
index 41de90f9f6..3df42ae448 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -201,8 +201,8 @@ netaddr = ">=0.7.18"
 # add a lower bound to the Jinja2 dependency.
 Jinja2 = ">=3.0"
 bleach = ">=1.4.3"
-# We use `Self`, which were added in `typing-extensions` 4.0.
-typing-extensions = ">=4.0"
+# We use `assert_never`, which were added in `typing-extensions` 4.1.
+typing-extensions = ">=4.1"
 # We enforce that we have a `cryptography` version that bundles an `openssl`
 # with the latest security patches.
 cryptography = ">=3.4.7"
diff --git a/synapse/handlers/sliding_sync.py b/synapse/handlers/sliding_sync.py
index c22a184efa..de067a87c6 100644
--- a/synapse/handlers/sliding_sync.py
+++ b/synapse/handlers/sliding_sync.py
@@ -610,6 +610,31 @@ class SlidingSyncHandler:
                 else:
                     relevant_room_map[room_id] = room_sync_config
 
+        # Filter out rooms that haven't received updates and we've sent down
+        # previously.
+        if from_token:
+            rooms_should_send = set()
+            for room_id in relevant_room_map:
+                status = await self.connection_store.have_sent_room(
+                    user_id,
+                    sync_config.connection_id(),
+                    from_token.connection_token,
+                    room_id,
+                )
+                if status.status != HaveSentRoomFlag.LIVE:
+                    rooms_should_send.add(room_id)
+
+            # TODO: Also check current state delta stream
+            rooms_that_have_updates = (
+                self.store._events_stream_cache.get_entities_changed(
+                    relevant_room_map, from_token.stream_token.room_key.stream
+                )
+            )
+            rooms_should_send.update(rooms_that_have_updates)
+            relevant_room_map = {
+                r: c for r, c in relevant_room_map.items() if r in rooms_should_send
+            }
+
         # Fetch room data
         rooms: Dict[str, SlidingSyncResult.RoomResult] = {}
 
diff --git a/tests/rest/client/test_sync.py b/tests/rest/client/test_sync.py
index b8d2a0a2d0..0c66111e07 100644
--- a/tests/rest/client/test_sync.py
+++ b/tests/rest/client/test_sync.py
@@ -3569,8 +3569,7 @@ class SlidingSyncTestCase(unittest.HomeserverTestCase):
 
         # We only return updates but only if we've sent the room down the
         # connection before.
-        self.assertIsNone(channel.json_body["rooms"][room_id1].get("required_state"))
-        self.assertIsNone(channel.json_body["rooms"][room_id1].get("invite_state"))
+        self.assertNotIn(room_id1, channel.json_body["rooms"])
 
     def test_rooms_required_state_wildcard(self) -> None:
         """
@@ -4323,6 +4322,187 @@ class SlidingSyncTestCase(unittest.HomeserverTestCase):
             channel.json_body["rooms"].get(room_id1), channel.json_body["rooms"]
         )
 
+    def test_incremental_sync_incremental_state(self) -> None:
+        """Test that we only get state updates in incremental sync for rooms
+        we've already seen.
+        """
+
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok)
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+
+        # Make the Sliding Sync request
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint,
+            {
+                "lists": {
+                    "foo-list": {
+                        "ranges": [[0, 1]],
+                        "required_state": [
+                            [EventTypes.Create, ""],
+                            [EventTypes.RoomHistoryVisibility, ""],
+                            # This one doesn't exist in the room
+                            [EventTypes.Name, ""],
+                        ],
+                        "timeline_limit": 0,
+                    }
+                }
+            },
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        from_token = channel.json_body["pos"]
+
+        state_map = self.get_success(
+            self.storage_controllers.state.get_current_state(room_id1)
+        )
+
+        self._assertRequiredStateIncludes(
+            channel.json_body["rooms"][room_id1]["required_state"],
+            {
+                state_map[(EventTypes.Create, "")],
+                state_map[(EventTypes.RoomHistoryVisibility, "")],
+            },
+            exact=True,
+        )
+
+        # Send a state event
+        self.helper.send_state(
+            room_id1, EventTypes.Name, body={"name": "foo"}, tok=user2_tok
+        )
+
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint + f"?pos={from_token}",
+            {
+                "lists": {
+                    "foo-list": {
+                        "ranges": [[0, 1]],
+                        "required_state": [
+                            [EventTypes.Create, ""],
+                            [EventTypes.RoomHistoryVisibility, ""],
+                            [EventTypes.Name, ""],
+                        ],
+                        "timeline_limit": 0,
+                    }
+                }
+            },
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        state_map = self.get_success(
+            self.storage_controllers.state.get_current_state(room_id1)
+        )
+
+        self._assertRequiredStateIncludes(
+            channel.json_body["rooms"][room_id1]["required_state"],
+            {
+                state_map[(EventTypes.Name, "")],
+            },
+            exact=True,
+        )
+
+    def test_incremental_sync_full_state_new_room(self) -> None:
+        """Test that we get state all state in incremental sync for rooms that
+        we haven't seen before.
+        """
+
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok)
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+
+        room_id2 = self.helper.create_room_as(user2_id, tok=user2_tok)
+        self.helper.join(room_id2, user1_id, tok=user1_tok)
+
+        # Make the Sliding Sync request, we'll only receive room_id2
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint,
+            {
+                "lists": {
+                    "foo-list": {
+                        "ranges": [[0, 0]],
+                        "required_state": [
+                            [EventTypes.Create, ""],
+                            [EventTypes.RoomHistoryVisibility, ""],
+                            # This one doesn't exist in the room
+                            [EventTypes.Name, ""],
+                        ],
+                        "timeline_limit": 0,
+                    }
+                }
+            },
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        from_token = channel.json_body["pos"]
+
+        state_map = self.get_success(
+            self.storage_controllers.state.get_current_state(room_id2)
+        )
+
+        self._assertRequiredStateIncludes(
+            channel.json_body["rooms"][room_id2]["required_state"],
+            {
+                state_map[(EventTypes.Create, "")],
+                state_map[(EventTypes.RoomHistoryVisibility, "")],
+            },
+            exact=True,
+        )
+        self.assertNotIn(room_id1, channel.json_body["rooms"])
+
+        # Send a state event in room 1
+        self.helper.send_state(
+            room_id1, EventTypes.Name, body={"name": "foo"}, tok=user2_tok
+        )
+
+        # We should get room_id1 down sync, with the full state.
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint + f"?pos={from_token}",
+            {
+                "lists": {
+                    "foo-list": {
+                        "ranges": [[0, 0]],
+                        "required_state": [
+                            [EventTypes.Create, ""],
+                            [EventTypes.RoomHistoryVisibility, ""],
+                            [EventTypes.Name, ""],
+                        ],
+                        "timeline_limit": 0,
+                    }
+                }
+            },
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        state_map = self.get_success(
+            self.storage_controllers.state.get_current_state(room_id1)
+        )
+
+        self._assertRequiredStateIncludes(
+            channel.json_body["rooms"][room_id1]["required_state"],
+            {
+                state_map[(EventTypes.Create, "")],
+                state_map[(EventTypes.RoomHistoryVisibility, "")],
+                state_map[(EventTypes.Name, "")],
+            },
+            exact=True,
+        )
+
 
 class SlidingSyncToDeviceExtensionTestCase(unittest.HomeserverTestCase):
     """Tests for the to-device sliding sync extension"""