Sliding sync: Correctly track which read receipts we have or have not sent down.

changelog.d/17574.misc

@@ -0,0 +1 @@
+Refactor per-connection state in experimental sliding sync handler.

changelog.d/17575.misc

@@ -0,0 +1 @@
+Correctly track read receipts that should be sent down in experimental sliding sync.

synapse/storage/databases/main/receipts.py

@@ -51,10 +51,12 @@
     JsonMapping,
     MultiWriterStreamToken,
     PersistedPosition,
+    StrCollection,
 )
 from synapse.util import json_encoder
 from synapse.util.caches.descriptors import cached, cachedList
 from synapse.util.caches.stream_change_cache import StreamChangeCache
+from synapse.util.iterutils import batch_iter
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -550,6 +552,46 @@ def f(txn: LoggingTransaction) -> List[Tuple[str, str, str, str, str]]:
 
         return results
 
+    async def get_rooms_with_receipts_between(
+        self,
+        room_ids: StrCollection,
+        from_key: MultiWriterStreamToken,
+        to_key: MultiWriterStreamToken,
+    ) -> StrCollection:
+        """Given a set of room_ids, find out which ones (may) have receipts
+        between the two tokens (> `from_token` and <= `to_token`)."""
+
+        room_ids = self._receipts_stream_cache.get_entities_changed(
+            room_ids, from_key.stream
+        )
+        if not room_ids:
+            return []
+
+        def f(txn: LoggingTransaction, room_ids: StrCollection) -> StrCollection:
+            clause, args = make_in_list_sql_clause(
+                self.database_engine, "room_id", room_ids
+            )
+
+            sql = f"""
+                SELECT DISTINCT room_id FROM receipts_linearized
+                WHERE {clause} AND ? < stream_id AND stream_id <= ?
+            """
+            args.append(from_key.stream)
+            args.append(to_key.get_max_stream_pos())
+
+            txn.execute(sql, args)
+
+            return [room_id for room_id, in txn]
+
+        results: List[str] = []
+        for batch in batch_iter(room_ids, 1000):
+            batch_result = await self.db_pool.runInteraction(
+                "get_rooms_with_receipts_between", f, batch
+            )
+            results.extend(batch_result)
+
+        return results
+
     async def get_users_sent_receipts_between(
         self, last_id: int, current_id: int
     ) -> List[str]:

tests/rest/client/sliding_sync/test_extension_receipts.py

@@ -677,3 +677,108 @@ def test_wait_for_new_data_timeout(self) -> None:
             set(),
             exact=True,
         )
+
+    def test_receipts_incremental_sync_out_of_range(self) -> None:
+        """Tests that we don't return read receipts for rooms that fall out of
+        range, but then do send all read receipts once they're back in range.
+        """
+
+        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)
+
+        # Send a message and read receipt into room2
+        event_response = self.helper.send(room_id2, body="new event", tok=user2_tok)
+        room2_event_id = event_response["event_id"]
+
+        self.helper.send_read_receipt(room_id2, room2_event_id, tok=user1_tok)
+
+        # Now send a message into room1 so that it is at the top of the list
+        self.helper.send(room_id1, body="new event", tok=user2_tok)
+
+        # Make a SS request for only the top room.
+        sync_body = {
+            "lists": {
+                "main": {
+                    "ranges": [[0, 0]],
+                    "required_state": [],
+                    "timeline_limit": 5,
+                }
+            },
+            "extensions": {
+                "receipts": {
+                    "enabled": True,
+                }
+            },
+        }
+        response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        # The receipt is in room2, but only room1 is returned, so we don't
+        # expect to get the receipt.
+        self.assertIncludes(
+            response_body["extensions"]["receipts"].get("rooms").keys(),
+            set(),
+            exact=True,
+        )
+
+        # Move room2 into range.
+        self.helper.send(room_id2, body="new event", tok=user2_tok)
+
+        response_body, from_token = self.do_sync(
+            sync_body, since=from_token, tok=user1_tok
+        )
+
+        # We expect to see the read receipt of room2, as that has the most
+        # recent update.
+        self.assertIncludes(
+            response_body["extensions"]["receipts"].get("rooms").keys(),
+            {room_id2},
+            exact=True,
+        )
+        receipt = response_body["extensions"]["receipts"]["rooms"][room_id2]
+        self.assertIncludes(
+            receipt["content"][room2_event_id][ReceiptTypes.READ].keys(),
+            {user1_id},
+            exact=True,
+        )
+
+        # Send a message into room1 to bump it to the top, but also send a
+        # receipt in room2
+        self.helper.send(room_id1, body="new event", tok=user2_tok)
+        self.helper.send_read_receipt(room_id2, room2_event_id, tok=user2_tok)
+
+        # We don't expect to see the new read receipt.
+        response_body, from_token = self.do_sync(
+            sync_body, since=from_token, tok=user1_tok
+        )
+        self.assertIncludes(
+            response_body["extensions"]["receipts"].get("rooms").keys(),
+            set(),
+            exact=True,
+        )
+
+        # But if we send a new message into room2, we expect to get the missing receipts
+        self.helper.send(room_id2, body="new event", tok=user2_tok)
+
+        response_body, from_token = self.do_sync(
+            sync_body, since=from_token, tok=user1_tok
+        )
+        self.assertIncludes(
+            response_body["extensions"]["receipts"].get("rooms").keys(),
+            {room_id2},
+            exact=True,
+        )
+
+        # We should only see the new receipt
+        receipt = response_body["extensions"]["receipts"]["rooms"][room_id2]
+        self.assertIncludes(
+            receipt["content"][room2_event_id][ReceiptTypes.READ].keys(),
+            {user2_id},
+            exact=True,
+        )

tests/rest/client/sliding_sync/test_extensions.py

@@ -120,19 +120,26 @@ def test_extensions_lists_rooms_relevant_rooms(
                 "foo-list": {
                     "ranges": [[0, 1]],
                     "required_state": [],
-                    "timeline_limit": 0,
+                    # We set this to `1` because we're testing `receipts` which
+                    # interact with the `timeline`. With receipts, when a room
+                    # hasn't been sent down the connection before or it appears
+                    # as `initial: true`, we only include receipts for events in
+                    # the timeline to avoid bloating and blowing up the sync
+                    # response as the number of users in the room increases.
+                    # (this behavior is part of the spec)
+                    "timeline_limit": 1,
                 },
                 # We expect this list range to include room5, room4, room3
                 "bar-list": {
                     "ranges": [[0, 2]],
                     "required_state": [],
-                    "timeline_limit": 0,
+                    "timeline_limit": 1,
                 },
             },
             "room_subscriptions": {
                 room_id1: {
                     "required_state": [],
-                    "timeline_limit": 0,
+                    "timeline_limit": 1,
                 }
             },
         }

tests/rest/client/utils.py

@@ -45,7 +45,7 @@
 from twisted.test.proto_helpers import MemoryReactorClock
 from twisted.web.server import Site
 
-from synapse.api.constants import Membership
+from synapse.api.constants import Membership, ReceiptTypes
 from synapse.api.errors import Codes
 from synapse.server import HomeServer
 from synapse.types import JsonDict
@@ -944,3 +944,15 @@ def initiate_sso_ui_auth(
         assert len(p.links) == 1, "not exactly one link in confirmation page"
         oauth_uri = p.links[0]
         return oauth_uri
+
+    def send_read_receipt(self, room_id: str, event_id: str, *, tok: str) -> None:
+        """Send a read receipt into the room at the given event"""
+        channel = make_request(
+            self.reactor,
+            self.site,
+            method="POST",
+            path=f"/rooms/{room_id}/receipt/{ReceiptTypes.READ}/{event_id}",
+            content={},
+            access_token=tok,
+        )
+        assert channel.code == HTTPStatus.OK, channel.text_body