Add stream_ordering sort to Sliding Sync /sync

changelog.d/17293.feature

@@ -0,0 +1 @@
+Add recency sort to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint.

changelog.d/17295.bugfix

@@ -0,0 +1 @@
+Fix edge case in `/sync` returning the wrong the state when using sharded event persisters.

synapse/handlers/sliding_sync.py

@@ -169,7 +169,7 @@ async def current_sync_for_user(
                 # `["m.room.member", "$LAZY"]`
                 filtered_room_ids = room_id_set
                 # TODO: Apply sorts
-                sorted_room_ids = sorted(filtered_room_ids)
+                sorted_room_ids = await self.sort_rooms(filtered_room_ids, to_token)
 
                 ops: List[SlidingSyncResult.SlidingWindowList.Operation] = []
                 if list_config.ranges:
@@ -439,3 +439,22 @@ async def get_sync_room_ids_for_user(
                 sync_room_id_set.add(room_id)
 
         return sync_room_id_set
+
+    async def sort_rooms(
+        self,
+        room_id_set: AbstractSet[str],
+        to_token: StreamToken,
+    ) -> List[str]:
+        """
+        Sort by recency of the last event in the room (stream_ordering). In order to get
+        a stable sort, we tie-break by room ID.
+
+        Args:
+            room_id_set: Set of room IDs to sort
+            to_token: We sort based on the events in the room at this token
+        """
+        # TODO: `get_last_event_in_room_before_stream_ordering()`
+
+        # TODO: Handle when people are left/banned from the room and shouldn't see past that point
+
+        return list(room_id_set)

synapse/storage/databases/main/stream.py

@@ -914,12 +914,23 @@ async def get_last_event_in_room_before_stream_ordering(
         def get_last_event_in_room_before_stream_ordering_txn(
             txn: LoggingTransaction,
         ) -> Optional[str]:
-            # We need to handle the fact that the stream tokens can be vector
-            # clocks. We do this by getting all rows between the minimum and
-            # maximum stream ordering in the token, plus one row less than the
-            # minimum stream ordering. We then filter the results against the
-            # token and return the first row that matches.
-
+            # We're looking for the closest event at or before the token. We need to
+            # handle the fact that the stream token can be a vector clock (with an
+            # `instance_map`) and events can be persisted on different instances
+            # (sharded event persisters). The first subquery handles the events that
+            # would be within the vector clock and gets all rows between the minimum and
+            # maximum stream ordering in the token which need to be filtered against the
+            # `instance_map`. The second subquery handles the "before" case and finds a
+            # row before the token. We then filter out any results past the token's
+            # vector clock and return the first row that matches.
+            min_stream = end_token.stream
+            max_stream = end_token.get_max_stream_pos()
+
+            # We use `union all` because we don't need any of the deduplication logic
+            # (`union` is really a union + distinct). `UNION ALL`` does preserve the
+            # ordering of the operand queries but there is no actual gurantee that it
+            # has this behavior in all scenarios so we need the extra `ORDER BY` at the
+            # bottom.
             sql = """
                 SELECT * FROM (
                     SELECT instance_name, stream_ordering, topological_ordering, event_id
@@ -931,7 +942,7 @@ def get_last_event_in_room_before_stream_ordering_txn(
                         AND rejections.event_id IS NULL
                     ORDER BY stream_ordering DESC
                 ) AS a
-                UNION
+                UNION ALL
                 SELECT * FROM (
                     SELECT instance_name, stream_ordering, topological_ordering, event_id
                     FROM events
@@ -943,15 +954,16 @@ def get_last_event_in_room_before_stream_ordering_txn(
                     ORDER BY stream_ordering DESC
                     LIMIT 1
                 ) AS b
+                ORDER BY stream_ordering DESC
             """
             txn.execute(
                 sql,
                 (
                     room_id,
-                    end_token.stream,
-                    end_token.get_max_stream_pos(),
+                    min_stream,
+                    max_stream,
                     room_id,
-                    end_token.stream,
+                    min_stream,
                 ),
             )
 

synapse/types/rest/client/__init__.py

@@ -175,20 +175,6 @@ class SlidingSyncList(CommonRoomParameters):
             ranges: Sliding window ranges. If this field is missing, no sliding window
                 is used and all rooms are returned in this list. Integers are
                 *inclusive*.
-            sort: How the list should be sorted on the server. The first value is
-                applied first, then tiebreaks are performed with each subsequent sort
-                listed.
-
-                    FIXME: Furthermore, it's not currently defined how servers should behave
-                    if they encounter a filter or sort operation they do not recognise. If
-                    the server rejects the request with an HTTP 400 then that will break
-                    backwards compatibility with new clients vs old servers. However, the
-                    client would be otherwise unaware that only some of the sort/filter
-                    operations have taken effect. We may need to include a "warnings"
-                    section to indicate which sort/filter operations are unrecognised,
-                    allowing for some form of graceful degradation of service.
-                    -- https://github.com/matrix-org/matrix-spec-proposals/blob/kegan/sync-v3/proposals/3575-sync.md#filter-and-sort-extensions
-
             slow_get_all_rooms: Just get all rooms (for clients that don't want to deal with
                 sliding windows). When true, the `ranges` and `sort` fields are ignored.
             required_state: Required state for each room returned. An array of event
@@ -253,7 +239,6 @@ class Filters(RequestBodyModel):
             ranges: Optional[List[Tuple[int, int]]] = None
         else:
             ranges: Optional[List[Tuple[conint(ge=0, strict=True), conint(ge=0, strict=True)]]] = None  # type: ignore[valid-type]
-        sort: Optional[List[StrictStr]] = None
         slow_get_all_rooms: Optional[StrictBool] = False
         include_heroes: Optional[StrictBool] = False
         filters: Optional[Filters] = None

tests/handlers/test_sliding_sync.py

@@ -1047,7 +1047,6 @@ def test_sharded_event_persisters(self) -> None:
 
         # Get a token while things are stuck after our activity
         stuck_activity_token = self.event_sources.get_current_token()
-        logger.info("stuck_activity_token %s", stuck_activity_token)
         # Let's make sure we're working with a token that has an `instance_map`
         self.assertNotEqual(len(stuck_activity_token.room_key.instance_map), 0)
 
@@ -1057,7 +1056,6 @@ def test_sharded_event_persisters(self) -> None:
         join_on_worker2_pos = self.get_success(
             self.store.get_position_for_event(join_on_worker2_response["event_id"])
         )
-        logger.info("join_on_worker2_pos %s", join_on_worker2_pos)
         # Ensure the join technially came after our token
         self.assertGreater(
             join_on_worker2_pos.stream,
@@ -1076,7 +1074,6 @@ def test_sharded_event_persisters(self) -> None:
         join_on_worker3_pos = self.get_success(
             self.store.get_position_for_event(join_on_worker3_response["event_id"])
         )
-        logger.info("join_on_worker3_pos %s", join_on_worker3_pos)
         # Ensure the join came after the min but still encapsulated by the token
         self.assertGreaterEqual(
             join_on_worker3_pos.stream,