Switch list quarantined media API to use max to_id instead of current

changelog.d/19677.feature

@@ -0,0 +1 @@
+Add a ["Listing quarantined media changes" Admin API](https://element-hq.github.io/synapse/latest/admin_api/media_admin_api.html#listing-quarantined-media-changes) for retrieving a paginated record of when media became (un)quarantined.

synapse/rest/admin/media.py

@@ -249,13 +249,22 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]:
 
         from_id = parse_integer(request, "from", default=0)
         limit = 100  # arbitrary; not enough to cause problems (hopefully)
-        to_id = await self.store.get_current_quarantined_media_stream_id()
 
-        if to_id < from_id:
-            # The caller is trying to get future data, which isn't possible.
+        # Validate the `from` token
+        max_id = await self.store.get_max_allocated_quarantined_media_stream_id()
+        if from_id > max_id:
+            # The caller is trying to get future data, which we don't allow because
+            # we know it's an invalid state that should never happen. We could
+            # wait until we reach the token but we might as well not waste our
+            # resources on that which is why `wait_for_quarantined_media_stream_id(...)`
+            # has assertions around this.
             raise SynapseError(
                 HTTPStatus.BAD_REQUEST,
-                "The `from` position is ahead of the currently persisted position.",
+                "The `from` token is considered invalid because it includes stream positions "
+                "greater than the furthest persisted position across all of the workers."
+                "This indicates either a Synapse programming error (as we should never hand out "
+                "invalid future tokens) or a fabricated `from` token. If you've modified the token, "
+                "you can try paginating from the beginning again.",
                 errcode=Codes.INVALID_PARAM,
             )
 
@@ -271,6 +280,7 @@ async def on_GET(self, request: SynapseRequest) -> tuple[int, JsonDict]:
                 errcode=Codes.UNKNOWN,
             )
 
+        to_id = await self.store.get_current_quarantined_media_stream_id()
         changes = await self.store.get_quarantined_media_changes(
             from_id=from_id,
             to_id=to_id,

synapse/storage/databases/main/room.py

@@ -1310,6 +1310,14 @@ async def get_current_quarantined_media_stream_id(self) -> int:
         """
         return self._quarantined_media_changes_id_gen.get_current_token()
 
+    async def get_max_allocated_quarantined_media_stream_id(self) -> int:
+        """Gets the maximum allocated position of the quarantined media changes stream.
+
+        Returns:
+            int - the maximum stream ID
+        """
+        return await self._quarantined_media_changes_id_gen.get_max_allocated_token()
+
     async def wait_for_quarantined_media_stream_id(self, target_id: int) -> bool:
         """Waits until the quarantined media changes stream reaches the given stream ID.