Switch list quarantined media API to use max to_id instead of current (#19677)

Following up on
https://github.com/element-hq/synapse/pull/19558#discussion_r3054831510

Changelog for this PR is intended to overlap with the above PR.

`get_current_quarantined_media_stream_id` wasn't being used anywhere
else, so we can replace it like we do in this PR.

### Pull Request Checklist

<!-- Please read
https://element-hq.github.io/synapse/latest/development/contributing_guide.html
before submitting your pull request -->

* [x] Pull request is based on the develop branch
* [x] Pull request includes a [changelog
file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog).
The entry should:
- Be a short description of your change which makes sense to users.
"Fixed a bug that prevented receiving messages from other servers."
instead of "Moved X method from `EventStore` to `EventWorkerStore`.".
  - Use markdown where necessary, mostly for `code blocks`.
  - End with either a period (.) or an exclamation mark (!).
  - Start with a capital letter.
- Feel free to credit yourself, by adding a sentence "Contributed by
@github_username." or "Contributed by [Your Name]." to the end of the
entry.
* [x] [Code
style](https://element-hq.github.io/synapse/latest/code_style.html) is
correct (run the
[linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters))

---------

Co-authored-by: Eric Eastwood <erice@element.io>
Co-authored-by: Eric Eastwood <madlittlemods@gmail.com>

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 @@ class ListQuarantineChanges(RestServlet):
 
         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 @@ class ListQuarantineChanges(RestServlet):
                 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 @@ class RoomWorkerStore(CacheInvalidationWorkerStore):
         """
         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.