diff --git a/changelog.d/19677.feature b/changelog.d/19677.feature new file mode 100644 index 00000000000..f4c8f042bf0 --- /dev/null +++ b/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. \ No newline at end of file diff --git a/synapse/rest/admin/media.py b/synapse/rest/admin/media.py index 5539b01af01..1633cca884a 100644 --- a/synapse/rest/admin/media.py +++ b/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, diff --git a/synapse/storage/databases/main/room.py b/synapse/storage/databases/main/room.py index 0a9e69e1bf5..8dbecc16e44 100644 --- a/synapse/storage/databases/main/room.py +++ b/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.