Add `rooms.bump_stamp` to Sliding Sync `/sync` for easier client-side sorting (#17395)

`bump_stamp` corresponds to the `stream_ordering` of the latest `DEFAULT_BUMP_EVENT_TYPES` in the room. This helps clients sort more readily without them needing to pull in a bunch of the timeline to determine the last activity. `bump_event_types` is a thing because for example, we don't want display name changes to mark the room as unread and bump it to the top. For encrypted rooms, we just have to consider any activity as a bump because we can't see the content and the client has to figure it out for themselves.

Outside of Synapse, `bump_stamp` is just a free-form counter so other implementations could use `received_ts`or `origin_server_ts` (see the [*Security considerations* section in MSC3575 about the potential pitfalls of using `origin_server_ts`](https://github.com/matrix-org/matrix-spec-proposals/blob/kegan/sync-v3/proposals/3575-sync.md#security-considerations)). It doesn't have any guarantee about always going up. In the Synapse case, it could go down if an event was redacted/removed (or purged in cases of retention policies).

In the future, we could add `bump_event_types` as [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) mentions if people need to customize the event types.

---

In the Sliding Sync proxy, a similar [`timestamp` field was added](https://github.com/matrix-org/sliding-sync/pull/247) for the same purpose but the name is not obvious what it pertains to or what it's for.

The `timestamp` field was also added to Ruma in https://github.com/ruma/ruma/pull/1622

3 files changed, 201 insertions, 4 deletions

diff --git a/tests/handlers/test_sliding_sync.py b/tests/handlers/test_sliding_sync.py
index 5f83b637c5..9dd2363adc 100644
--- a/tests/handlers/test_sliding_sync.py
+++ b/tests/handlers/test_sliding_sync.py
@@ -2844,7 +2844,7 @@ class SortRoomsTestCase(HomeserverTestCase):
         )
 
         # Sort the rooms (what we're testing)
-        sorted_room_info = self.get_success(
+        sorted_sync_rooms = self.get_success(
             self.sliding_sync_handler.sort_rooms(
                 sync_room_map=sync_room_map,
                 to_token=after_rooms_token,
@@ -2852,7 +2852,7 @@ class SortRoomsTestCase(HomeserverTestCase):
         )
 
         self.assertEqual(
-            [room_id for room_id, _ in sorted_room_info],
+            [room_membership.room_id for room_membership in sorted_sync_rooms],
             [room_id2, room_id1],
         )
 
@@ -2927,7 +2927,7 @@ class SortRoomsTestCase(HomeserverTestCase):
         )
 
         # Sort the rooms (what we're testing)
-        sorted_room_info = self.get_success(
+        sorted_sync_rooms = self.get_success(
             self.sliding_sync_handler.sort_rooms(
                 sync_room_map=sync_room_map,
                 to_token=after_rooms_token,
@@ -2935,7 +2935,7 @@ class SortRoomsTestCase(HomeserverTestCase):
         )
 
         self.assertEqual(
-            [room_id for room_id, _ in sorted_room_info],
+            [room_membership.room_id for room_membership in sorted_sync_rooms],
             [room_id2, room_id1, room_id3],
             "Corresponding map to disambiguate the opaque room IDs: "
             + str(
@@ -2946,3 +2946,63 @@ class SortRoomsTestCase(HomeserverTestCase):
                 }
             ),
         )
+
+    def test_default_bump_event_types(self) -> None:
+        """
+        Test that we only consider the *latest* event in the room when sorting (not
+        `bump_event_types`).
+        """
+        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,
+        )
+        message_response = self.helper.send(room_id1, "message in room1", tok=user1_tok)
+        room_id2 = self.helper.create_room_as(
+            user1_id,
+            tok=user1_tok,
+        )
+        self.helper.send(room_id2, "message in room2", tok=user1_tok)
+
+        # Send a reaction in room1 which isn't in `DEFAULT_BUMP_EVENT_TYPES` but we only
+        # care about sorting by the *latest* event in the room.
+        self.helper.send_event(
+            room_id1,
+            type=EventTypes.Reaction,
+            content={
+                "m.relates_to": {
+                    "event_id": message_response["event_id"],
+                    "key": "👍",
+                    "rel_type": "m.annotation",
+                }
+            },
+            tok=user1_tok,
+        )
+
+        after_rooms_token = self.event_sources.get_current_token()
+
+        # Get the rooms the user should be syncing with
+        sync_room_map = self.get_success(
+            self.sliding_sync_handler.get_sync_room_ids_for_user(
+                UserID.from_string(user1_id),
+                from_token=None,
+                to_token=after_rooms_token,
+            )
+        )
+
+        # Sort the rooms (what we're testing)
+        sorted_sync_rooms = self.get_success(
+            self.sliding_sync_handler.sort_rooms(
+                sync_room_map=sync_room_map,
+                to_token=after_rooms_token,
+            )
+        )
+
+        self.assertEqual(
+            [room_membership.room_id for room_membership in sorted_sync_rooms],
+            # room1 sorts before room2 because it has the latest event (the reaction).
+            # We only care about the *latest* event in the room.
+            [room_id1, room_id2],
+        )
diff --git a/tests/rest/client/test_sync.py b/tests/rest/client/test_sync.py
index cb2888409e..6ff1f03c9a 100644
--- a/tests/rest/client/test_sync.py
+++ b/tests/rest/client/test_sync.py
@@ -2029,6 +2029,102 @@ class SlidingSyncTestCase(unittest.HomeserverTestCase):
             channel.json_body["rooms"][room_id1],
         )
 
+    def test_rooms_bump_stamp(self) -> None:
+        """
+        Test that `bump_stamp` is present and pointing to relevant events.
+        """
+        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,
+        )
+        event_response1 = message_response = self.helper.send(
+            room_id1, "message in room1", tok=user1_tok
+        )
+        event_pos1 = self.get_success(
+            self.store.get_position_for_event(event_response1["event_id"])
+        )
+        room_id2 = self.helper.create_room_as(
+            user1_id,
+            tok=user1_tok,
+        )
+        send_response2 = self.helper.send(room_id2, "message in room2", tok=user1_tok)
+        event_pos2 = self.get_success(
+            self.store.get_position_for_event(send_response2["event_id"])
+        )
+
+        # Send a reaction in room1 but it shouldn't affect the `bump_stamp`
+        # because reactions are not part of the `DEFAULT_BUMP_EVENT_TYPES`
+        self.helper.send_event(
+            room_id1,
+            type=EventTypes.Reaction,
+            content={
+                "m.relates_to": {
+                    "event_id": message_response["event_id"],
+                    "key": "👍",
+                    "rel_type": "m.annotation",
+                }
+            },
+            tok=user1_tok,
+        )
+
+        # Make the Sliding Sync request
+        timeline_limit = 100
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint,
+            {
+                "lists": {
+                    "foo-list": {
+                        "ranges": [[0, 1]],
+                        "required_state": [],
+                        "timeline_limit": timeline_limit,
+                    }
+                }
+            },
+            access_token=user1_tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        # Make sure it has the foo-list we requested
+        self.assertListEqual(
+            list(channel.json_body["lists"].keys()),
+            ["foo-list"],
+            channel.json_body["lists"].keys(),
+        )
+
+        # Make sure the list includes the rooms in the right order
+        self.assertListEqual(
+            list(channel.json_body["lists"]["foo-list"]["ops"]),
+            [
+                {
+                    "op": "SYNC",
+                    "range": [0, 1],
+                    # room1 sorts before room2 because it has the latest event (the
+                    # reaction)
+                    "room_ids": [room_id1, room_id2],
+                }
+            ],
+            channel.json_body["lists"]["foo-list"],
+        )
+
+        # The `bump_stamp` for room1 should point at the latest message (not the
+        # reaction since it's not one of the `DEFAULT_BUMP_EVENT_TYPES`)
+        self.assertEqual(
+            channel.json_body["rooms"][room_id1]["bump_stamp"],
+            event_pos1.stream,
+            channel.json_body["rooms"][room_id1],
+        )
+
+        # The `bump_stamp` for room2 should point at the latest message
+        self.assertEqual(
+            channel.json_body["rooms"][room_id2]["bump_stamp"],
+            event_pos2.stream,
+            channel.json_body["rooms"][room_id2],
+        )
+
     def test_rooms_newly_joined_incremental_sync(self) -> None:
         """
         Test that when we make an incremental sync with a `newly_joined` `rooms`, we are
diff --git a/tests/storage/test_stream.py b/tests/storage/test_stream.py
index aad46b1b44..9dea1af8ea 100644
--- a/tests/storage/test_stream.py
+++ b/tests/storage/test_stream.py
@@ -556,6 +556,47 @@ class GetLastEventInRoomBeforeStreamOrderingTestCase(HomeserverTestCase):
             ),
         )
 
+    def test_restrict_event_types(self) -> None:
+        """
+        Test that we only consider given `event_types` when finding the last event
+        before a token.
+        """
+        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, is_public=True)
+        event_response = self.helper.send_event(
+            room_id1,
+            type="org.matrix.special_message",
+            content={"body": "before1, target!"},
+            tok=user1_tok,
+        )
+        self.helper.send(room_id1, "before2", tok=user1_tok)
+
+        after_room_token = self.event_sources.get_current_token()
+
+        # Send some events after the token
+        self.helper.send_event(
+            room_id1,
+            type="org.matrix.special_message",
+            content={"body": "after1"},
+            tok=user1_tok,
+        )
+        self.helper.send(room_id1, "after2", tok=user1_tok)
+
+        last_event_result = self.get_success(
+            self.store.get_last_event_pos_in_room_before_stream_ordering(
+                room_id=room_id1,
+                end_token=after_room_token.room_key,
+                event_types=["org.matrix.special_message"],
+            )
+        )
+        assert last_event_result is not None
+        last_event_id, _ = last_event_result
+
+        # Make sure it's the last event before the token
+        self.assertEqual(last_event_id, event_response["event_id"])
+
 
 class GetCurrentStateDeltaMembershipChangesForUserTestCase(HomeserverTestCase):
     """