Add log to determine whether clients are using /messages as expected (#19226)

Spawning from wanting some better homeserver logs to debug
#19153. We can check whether
we are returning a `/messages` response with an `end` pagination token
and then check to see whether the client is making another `/messages`
request with that token.

Although clients should also have similar logs and debugging
capabilities to determine this info as well. This just makes it easier
for us when someone creates an issue claiming backend issue and we can
ask them for homeserver logs.

Author: MadLittleMods

changelog.d/19226.misc

@@ -0,0 +1 @@
+Add log to determine whether clients are using `/messages` as expected.

synapse/handlers/pagination.py

@@ -21,22 +21,25 @@
 import logging
 from typing import TYPE_CHECKING, cast
 
+import attr
+
 from twisted.python.failure import Failure
 
 from synapse.api.constants import Direction, EventTypes, Membership
 from synapse.api.errors import SynapseError
 from synapse.api.filtering import Filter
-from synapse.events.utils import SerializeEventConfig
+from synapse.events import EventBase
+from synapse.handlers.relations import BundledAggregations
 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.streams.config import PaginationConfig
 from synapse.types import (
-    JsonDict,
     JsonMapping,
     Requester,
     ScheduledTask,
     StreamKeyType,
+    StreamToken,
     TaskStatus,
 )
 from synapse.types.handlers import ShutdownRoomParams, ShutdownRoomResponse
@@ -70,6 +73,58 @@
 SHUTDOWN_AND_PURGE_ROOM_ACTION_NAME = "shutdown_and_purge_room"
 
 
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class GetMessagesResult:
+    """
+    Everything needed to serialize a `/messages` response.
+    """
+
+    messages_chunk: list[EventBase]
+    """
+    A list of room events.
+
+     - When the request is `Direction.FORWARDS`, events will be in the range:
+       `start_token` < x <= `end_token`, (ascending topological_order)
+     - When the request is `Direction.BACKWARDS`, events will be in the range:
+       `start_token` >= x > `end_token`, (descending topological_order)
+
+    Note that an empty chunk does not necessarily imply that no more events are
+    available. Clients should continue to paginate until no `end_token` property is returned.
+    """
+
+    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
+    """
+    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
+    events for the senders of events in the chunk.
+
+    Omitted from the response when `None`.
+    """
+
+    start_token: StreamToken
+    """
+    Token corresponding to the start of chunk. This will be the same as the value given
+    in `from` query parameter of the `/messages` request.
+    """
+
+    end_token: StreamToken | None
+    """
+    A token corresponding to the end of chunk. This token can be passed back to this
+    endpoint to request further events.
+
+    If no further events are available (either because we have reached the start of the
+    timeline, or because the user does not have permission to see any more events), this
+    property is omitted from the response.
+    """
+
+
 class PaginationHandler:
     """Handles pagination and purge history requests.
 
@@ -418,7 +473,7 @@ async def get_messages(
         as_client_event: bool = True,
         event_filter: Filter | None = None,
         use_admin_priviledge: bool = False,
-    ) -> JsonDict:
+    ) -> GetMessagesResult:
         """Get messages in a room.
 
         Args:
@@ -617,10 +672,13 @@ async def get_messages(
         # In that case we do not return end, to tell the client
         # there is no need for further queries.
         if not events:
-            return {
-                "chunk": [],
-                "start": await from_token.to_string(self.store),
-            }
+            return GetMessagesResult(
+                messages_chunk=[],
+                bundled_aggregations={},
+                state=None,
+                start_token=from_token,
+                end_token=None,
+            )
 
         if event_filter:
             events = await event_filter.filter(events)
@@ -636,11 +694,13 @@ async def get_messages(
         # if after the filter applied there are no more events
         # return immediately - but there might be more in next_token batch
         if not events:
-            return {
-                "chunk": [],
-                "start": await from_token.to_string(self.store),
-                "end": await next_token.to_string(self.store),
-            }
+            return GetMessagesResult(
+                messages_chunk=[],
+                bundled_aggregations={},
+                state=None,
+                start_token=from_token,
+                end_token=next_token,
+            )
 
         state = None
         if event_filter and event_filter.lazy_load_members and len(events) > 0:
@@ -657,38 +717,20 @@ async def get_messages(
 
             if state_ids:
                 state_dict = await self.store.get_events(list(state_ids.values()))
-                state = state_dict.values()
+                state = list(state_dict.values())
 
         aggregations = await self._relations_handler.get_bundled_aggregations(
             events, user_id
         )
 
-        time_now = self.clock.time_msec()
-
-        serialize_options = SerializeEventConfig(
-            as_client_event=as_client_event, requester=requester
+        return GetMessagesResult(
+            messages_chunk=events,
+            bundled_aggregations=aggregations,
+            state=state,
+            start_token=from_token,
+            end_token=next_token,
         )
 
-        chunk = {
-            "chunk": (
-                await self._event_serializer.serialize_events(
-                    events,
-                    time_now,
-                    config=serialize_options,
-                    bundle_aggregations=aggregations,
-                )
-            ),
-            "start": await from_token.to_string(self.store),
-            "end": await next_token.to_string(self.store),
-        }
-
-        if state:
-            chunk["state"] = await self._event_serializer.serialize_events(
-                state, time_now, config=serialize_options
-            )
-
-        return chunk
-
     async def _shutdown_and_purge_room(
         self,
         task: ScheduledTask,

synapse/rest/admin/rooms.py

@@ -28,9 +28,13 @@
 from synapse.api.constants import Direction, EventTypes, JoinRules, Membership
 from synapse.api.errors import AuthError, Codes, NotFoundError, SynapseError
 from synapse.api.filtering import Filter
+from synapse.events.utils import (
+    SerializeEventConfig,
+)
 from synapse.handlers.pagination import (
     PURGE_ROOM_ACTION_NAME,
     SHUTDOWN_AND_PURGE_ROOM_ACTION_NAME,
+    GetMessagesResult,
 )
 from synapse.http.servlet import (
     ResolveRoomIdMixin,
@@ -44,11 +48,13 @@
     parse_string,
 )
 from synapse.http.site import SynapseRequest
+from synapse.logging.opentracing import trace
 from synapse.rest.admin._base import (
     admin_patterns,
     assert_requester_is_admin,
     assert_user_is_admin,
 )
+from synapse.rest.client.room import SerializeMessagesDeps, encode_messages_response
 from synapse.storage.databases.main.room import RoomSortOrder
 from synapse.streams.config import PaginationConfig
 from synapse.types import JsonDict, RoomID, ScheduledTask, UserID, create_requester
@@ -976,6 +982,7 @@ def __init__(self, hs: "HomeServer"):
         self._pagination_handler = hs.get_pagination_handler()
         self._auth = hs.get_auth()
         self._store = hs.get_datastores().main
+        self._event_serializer = hs.get_event_client_serializer()
 
     async def on_GET(
         self, request: SynapseRequest, room_id: str
@@ -999,7 +1006,11 @@ async def on_GET(
         ):
             as_client_event = False
 
-        msgs = await self._pagination_handler.get_messages(
+        serialize_options = SerializeEventConfig(
+            as_client_event=as_client_event, requester=requester
+        )
+
+        get_messages_result = await self._pagination_handler.get_messages(
             room_id=room_id,
             requester=requester,
             pagin_config=pagination_config,
@@ -1008,7 +1019,27 @@ async def on_GET(
             use_admin_priviledge=True,
         )
 
-        return HTTPStatus.OK, msgs
+        response_content = await self.encode_response(
+            get_messages_result, serialize_options
+        )
+
+        return HTTPStatus.OK, response_content
+
+    @trace
+    async def encode_response(
+        self,
+        get_messages_result: GetMessagesResult,
+        serialize_options: SerializeEventConfig,
+    ) -> JsonDict:
+        return await encode_messages_response(
+            get_messages_result=get_messages_result,
+            serialize_options=serialize_options,
+            serialize_deps=SerializeMessagesDeps(
+                clock=self._clock,
+                event_serializer=self._event_serializer,
+                store=self._store,
+            ),
+        )
 
 
 class RoomTimestampToEventRestServlet(RestServlet):