Implement MSC3871: Gappy timelines - indicate gaps in /messages

changelog.d/18873.feature

@@ -0,0 +1 @@
+Implement experimental [MSC3871](https://github.com/matrix-org/matrix-spec-proposals/pull/3871) to indicate `gaps` in the `/messages` timeline.

synapse/handlers/federation.py

@@ -253,7 +253,9 @@ async def _maybe_backfill_inner(
             _BackfillPoint(event_id, depth, _BackfillPointType.BACKWARDS_EXTREMITY)
             for event_id, depth in await self.store.get_backfill_points_in_room(
                 room_id=room_id,
-                current_depth=current_depth,
+                # Per the docstring, it's best to pad the `current_depth` by the
+                # number of messages you plan to backfill from these points.
+                current_depth=current_depth + limit,
                 # We only need to end up with 5 extremities combined with the
                 # insertion event extremities to make the `/backfill` request
                 # but fetch an order of magnitude more to make sure there is

synapse/handlers/pagination.py

@@ -19,7 +19,7 @@
 #
 #
 import logging
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING, Sequence, cast
 
 import attr
 
@@ -33,6 +33,7 @@
 from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME
 from synapse.logging.opentracing import trace
 from synapse.rest.admin._base import assert_user_is_admin
+from synapse.storage.databases.main.events_worker import EventGapEntry
 from synapse.streams.config import PaginationConfig
 from synapse.types import (
     JsonMapping,
@@ -79,7 +80,7 @@ class GetMessagesResult:
     Everything needed to serialize a `/messages` response.
     """
 
-    messages_chunk: list[EventBase]
+    messages_chunk: Sequence[EventBase]
     """
     A list of room events.
 
@@ -92,14 +93,19 @@ class GetMessagesResult:
     available. Clients should continue to paginate until no `end_token` property is returned.
     """
 
+    gaps: Sequence[EventGapEntry]
+    """
+    A list of gaps in the `messages_chunk`
+    """
+
     bundled_aggregations: dict[str, BundledAggregations]
     """
     A map of event ID to the bundled aggregations for the events in the chunk.
 
     If an event doesn't have any bundled aggregations, it may not appear in the map.
     """
 
-    state: list[EventBase] | None
+    state: Sequence[EventBase] | None
     """
     A list of state events relevant to showing the chunk. For example, if
     lazy_load_members is enabled in the filter then this may contain the membership
@@ -467,12 +473,14 @@ async def purge_room(
     @trace
     async def get_messages(
         self,
+        *,
         requester: Requester,
         room_id: str,
         pagin_config: PaginationConfig,
         as_client_event: bool = True,
         event_filter: Filter | None = None,
         use_admin_priviledge: bool = False,
+        backfill: bool = True,
     ) -> GetMessagesResult:
         """Get messages in a room.
 
@@ -485,6 +493,8 @@ async def get_messages(
             use_admin_priviledge: if `True`, return all events, regardless
                 of whether `user` has access to them. To be used **ONLY**
                 from the admin API.
+            backfill: If false, we skip backfill altogether. When true, we backfill as a
+                best effort.
 
         Returns:
             Pagination API results
@@ -575,7 +585,7 @@ async def get_messages(
             event_filter=event_filter,
         )
 
-        if pagin_config.direction == Direction.BACKWARDS:
+        if backfill and pagin_config.direction == Direction.BACKWARDS:
             # We use a `Set` because there can be multiple events at a given depth
             # and we only care about looking at the unique continum of depths to
             # find gaps.
@@ -674,6 +684,7 @@ async def get_messages(
         if not events:
             return GetMessagesResult(
                 messages_chunk=[],
+                gaps=[],
                 bundled_aggregations={},
                 state=None,
                 start_token=from_token,
@@ -696,6 +707,7 @@ async def get_messages(
         if not events:
             return GetMessagesResult(
                 messages_chunk=[],
+                gaps=[],
                 bundled_aggregations={},
                 state=None,
                 start_token=from_token,
@@ -723,8 +735,13 @@ async def get_messages(
             events, user_id
         )
 
+        gaps = await self.store.get_events_next_to_gaps(
+            events=events, direction=pagin_config.direction
+        )
+
         return GetMessagesResult(
             messages_chunk=events,
+            gaps=gaps,
             bundled_aggregations=aggregations,
             state=state,
             start_token=from_token,

synapse/rest/client/room.py

@@ -79,7 +79,14 @@
 from synapse.state import CREATE_KEY, POWER_KEY
 from synapse.storage.databases.main import DataStore
 from synapse.streams.config import PaginationConfig
-from synapse.types import JsonDict, Requester, StreamToken, ThirdPartyInstanceID, UserID
+from synapse.types import (
+    JsonDict,
+    Requester,
+    StreamKeyType,
+    StreamToken,
+    ThirdPartyInstanceID,
+    UserID,
+)
 from synapse.types.state import StateFilter
 from synapse.util.cancellation import cancellable
 from synapse.util.clock import Clock
@@ -854,6 +861,18 @@ async def encode_messages_response(
                 bundle_aggregations=get_messages_result.bundled_aggregations,
             )
         ),
+        "org.matrix.msc3871.gaps": [
+            {
+                "prev_pagination_token": await get_messages_result.start_token.copy_and_replace(
+                    StreamKeyType.ROOM, gap.prev_token
+                ).to_string(serialize_deps.store),
+                "event_id": gap.event_id,
+                "next_pagination_token": await get_messages_result.start_token.copy_and_replace(
+                    StreamKeyType.ROOM, gap.next_token
+                ).to_string(serialize_deps.store),
+            }
+            for gap in get_messages_result.gaps
+        ],
         "start": await get_messages_result.start_token.to_string(serialize_deps.store),
     }
 
@@ -893,6 +912,16 @@ def __init__(self, hs: "HomeServer"):
     async def on_GET(
         self, request: SynapseRequest, room_id: str
     ) -> tuple[int, JsonDict]:
+        """
+        Query paremeters:
+            dir
+            from
+            to
+            limit
+            filter
+            backfill: If false, we skip backfill altogether. When true, we backfill as a
+                best effort.
+        """
         processing_start_time = self.clock.time_msec()
         # Fire off and hope that we get a result by the end.
         #
@@ -922,6 +951,8 @@ async def on_GET(
         ):
             as_client_event = False
 
+        backfill = parse_boolean(request, "backfill", default=True)
+
         serialize_options = SerializeEventConfig(
             as_client_event=as_client_event, requester=requester
         )
@@ -932,6 +963,7 @@ async def on_GET(
             pagin_config=pagination_config,
             as_client_event=as_client_event,
             event_filter=event_filter,
+            backfill=backfill,
         )
 
         # Useful for debugging timeline/pagination issues. For example, if a client

synapse/storage/databases/main/event_federation.py

@@ -1210,6 +1210,30 @@ async def get_backfill_points_in_room(
         equal to the `current_depth`. Sorted by depth, highest to lowest (descending)
         so the closest events to the `current_depth` are first in the list.
 
+        Note: We can only do approximate depth comparisons. Backwards extremeties are
+        the oldest events we know of in the room but we only know of them because some
+        other event referenced them by prev_event and aren't persisted in our database
+        yet (meaning we don't know their depth specifically). So we need to look for the
+        approximate depth from the events connected to the current backwards
+        extremeties.
+
+        It's best to pad the `current_depth` by the number of messages you plan to
+        backfill from these points.
+
+        Example:
+
+         - Your pagination token represents a scroll position at `depth` of `100`.
+         - We have a backfill point at an approximate depth of `125`
+         - You plan to backfill `50` events from that backfill point.
+
+        When we pad our `current_depth`, `100` + `50` = `150`, we pick up the backfill
+        point at `125` (because <= `150`, our `current_depth`), backfill `50` events to
+        a depth of `75` in the timeline (exposing new events that we can return `100` ->
+        `75`).
+
+        When we don't pad our `current_depth`, `100` is lower than any of the backfill
+        points so we don't pick any and miss out on backfilling any events.
+
         We ignore extremities that are newer than the user's current scroll position
         (ie, those with depth greater than `current_depth`) as:
             1. we don't really care about getting events that have happened
@@ -1221,7 +1245,7 @@ async def get_backfill_points_in_room(
 
         Args:
             room_id: Room where we want to find the oldest events
-            current_depth: The depth at the user's current scrollback position
+            current_depth: The depth at the user's current scrollback position (see notes above).
             limit: The max number of backfill points to return
 
         Returns: