Wake rooms into the page for receipt and account-data updates

Candidates were derived purely from the event stream, so a read receipt
(or room account data change) in an otherwise quiet room was deferred
until someone spoke in it. Rooms with undelivered receipt/account-data
changes - tracked PREVIOUSLY on the per-connection stream maps, or with
activity in the token range - now count as candidates when the relevant
extension is enabled: the room wakes into the page (usually as an empty,
filtered-out entry) and the extension delivers the data.

synapse/handlers/sliding_sync/paginated.py

@@ -289,6 +289,61 @@ class PaginatedSyncHandler(SlidingSyncHandler):
             candidates.update(newly_joined_rooms & sync_room_map.keys())
             candidates.update(newly_left_rooms & sync_room_map.keys())
 
+            # Rooms with undelivered receipt or room account data changes must
+            # also wake into the page, else a read receipt in an otherwise
+            # quiet room is deferred until someone speaks. The room comes down
+            # as an (often empty, filtered-out) entry and the extension
+            # delivers the data. Only streams the client has enabled can have
+            # anything to deliver.
+            extensions_body = sync_config.extensions
+            if (
+                extensions_body is not None
+                and extensions_body.receipts is not None
+                and extensions_body.receipts.enabled
+            ):
+                # Rooms already recorded as having undelivered receipts, plus
+                # rooms with new receipt activity in the token range.
+                candidates.update(
+                    room_id
+                    for room_id, receipt_status in previous_connection_state.receipts._statuses.items()
+                    if receipt_status.status == HaveSentRoomFlag.PREVIOUSLY
+                    and room_id in sync_room_map
+                )
+                candidates.update(
+                    await self.store.get_rooms_with_receipts_between(
+                        [
+                            room_id
+                            for room_id in sync_room_map
+                            if room_id not in candidates
+                        ],
+                        from_key=from_token.stream_token.receipt_key,
+                        to_key=to_token.receipt_key,
+                    )
+                )
+            if (
+                extensions_body is not None
+                and extensions_body.account_data is not None
+                and extensions_body.account_data.enabled
+            ):
+                candidates.update(
+                    room_id
+                    for room_id, account_data_status in previous_connection_state.account_data._statuses.items()
+                    if account_data_status.status == HaveSentRoomFlag.PREVIOUSLY
+                    and room_id in sync_room_map
+                )
+                updated_account_data = (
+                    await self.store.get_updated_room_account_data_for_user(
+                        user_id, from_token.stream_token.account_data_key
+                    )
+                )
+                updated_tags = await self.store.get_updated_tags(
+                    user_id, from_token.stream_token.account_data_key
+                )
+                candidates.update(
+                    (updated_account_data.keys() | updated_tags.keys())
+                    & sync_room_map.keys()
+                )
+
         # Page: most recently active rooms first. When the page overflows, a
         # slice of it is reserved for the longest-deferred rooms so that
         # nothing is starved by busier rooms perpetually sorting first.

tests/rest/client/sliding_sync/test_paginated_sync.py

@@ -21,7 +21,14 @@ from unittest.mock import AsyncMock
 from twisted.internet.testing import MemoryReactor
 
 import synapse.rest.admin
-from synapse.rest.client import login, paginated_sync, receipts, room, sync
+from synapse.rest.client import (
+    account_data,
+    login,
+    paginated_sync,
+    receipts,
+    room,
+    sync,
+)
 from synapse.server import HomeServer
 from synapse.types import JsonDict
 from synapse.util.clock import Clock
@@ -40,6 +47,7 @@ class PaginatedSyncTestCase(unittest.HomeserverTestCase):
 
     servlets = [
         synapse.rest.admin.register_servlets,
+        account_data.register_servlets,
         login.register_servlets,
         receipts.register_servlets,
         room.register_servlets,
@@ -336,3 +344,67 @@ class PaginatedSyncTestCase(unittest.HomeserverTestCase):
         self.assertIn(room_id, response["rooms"])
         receipts_response = response["extensions"]["receipts"]["rooms"]
         self.assertIn(room_id, receipts_response, receipts_response)
+
+    def test_receipt_in_quiet_room_wakes_room(self) -> None:
+        """A read receipt in a room with no new events must still be
+        delivered: the room is woken into the page and the receipts extension
+        carries the receipt (the room entry itself stays empty and is
+        filtered out)."""
+        room_id = self.helper.create_room_as(self.user, tok=self.tok)
+        user2 = self.register_user("bob", "password")
+        tok2 = self.login("bob", "password")
+        self.helper.join(room_id, user2, tok=tok2)
+        event_response = self.helper.send(room_id, body="hello", tok=self.tok)
+
+        body = {
+            "page_size": 10,
+            "limit": 5,
+            "history": 5,
+            "extensions": {"receipts": {"enabled": True}},
+        }
+        response = self._sync(body)
+        pos = response["pos"]
+
+        # Bob reads the room; no events are sent anywhere.
+        channel = self.make_request(
+            "POST",
+            f"/rooms/{room_id}/receipt/m.read/{event_response['event_id']}",
+            {},
+            access_token=tok2,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        response = self._sync(body, pos=pos)
+        receipts_response = response["extensions"]["receipts"]["rooms"]
+        self.assertIn(room_id, receipts_response, response)
+        # The room itself had nothing to say.
+        self.assertNotIn(room_id, response["rooms"])
+
+    def test_account_data_in_quiet_room_wakes_room(self) -> None:
+        """Room account data (e.g. read-state set from another device) in a
+        room with no new events must still be delivered via the account_data
+        extension."""
+        room_id = self.helper.create_room_as(self.user, tok=self.tok)
+        self.helper.send(room_id, body="hello", tok=self.tok)
+
+        body = {
+            "page_size": 10,
+            "limit": 5,
+            "history": 5,
+            "extensions": {"account_data": {"enabled": True}},
+        }
+        response = self._sync(body)
+        pos = response["pos"]
+
+        # "Another device" updates the room's account data; no events anywhere.
+        channel = self.make_request(
+            "PUT",
+            f"/user/{self.user}/rooms/{room_id}/account_data/org.example.read_state",
+            {"event_id": "$dummy"},
+            access_token=self.tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        response = self._sync(body, pos=pos)
+        account_data_response = response["extensions"]["account_data"]["rooms"]
+        self.assertIn(room_id, account_data_response, response)