From 33b05210b90e37888d243c2bc5ca2a6425cbd745 Mon Sep 17 00:00:00 2001 From: Andrew Morgan Date: Wed, 15 Jul 2026 15:36:48 +0100 Subject: [PATCH] initial llm implementation of device management linearization Intended to address cases where lots of device deletions for a single user come in at once. This may be better solved by MAS batching up said deletions (somehow). Need to chat with the MAS folks whether this is the right direction/a good stepping stone at the very least. --- synapse/handlers/device.py | 30 +++-- synapse/rest/synapse/mas/devices.py | 184 +++++++++++++++------------- 2 files changed, 113 insertions(+), 101 deletions(-) diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py index 22254666487..f22c9f4765c 100644 --- a/synapse/handlers/device.py +++ b/synapse/handlers/device.py @@ -139,6 +139,9 @@ def __init__(self, hs: "HomeServer"): hs.config.registration.dont_notify_new_devices_for ) + self.device_management_linearizer = Linearizer( + name="device_management", clock=self.clock + ) self.device_list_updater = DeviceListWorkerUpdater(hs) self._task_scheduler.register_action( @@ -890,21 +893,22 @@ async def _delete_device_messages( device_id = task.params["device_id"] up_to_stream_id = task.params["up_to_stream_id"] - # Delete the messages in batches to avoid too much DB load. - from_stream_id = None - while True: - from_stream_id, _ = await self.store.delete_messages_for_device_between( - user_id=user_id, - device_id=device_id, - from_stream_id=from_stream_id, - to_stream_id=up_to_stream_id, - limit=DeviceWriterHandler.DEVICE_MSGS_DELETE_BATCH_LIMIT, - ) + async with self.device_management_linearizer.queue(user_id): + # Delete the messages in batches to avoid too much DB load. + from_stream_id = None + while True: + from_stream_id, _ = await self.store.delete_messages_for_device_between( + user_id=user_id, + device_id=device_id, + from_stream_id=from_stream_id, + to_stream_id=up_to_stream_id, + limit=DeviceWriterHandler.DEVICE_MSGS_DELETE_BATCH_LIMIT, + ) - if from_stream_id is None: - return TaskStatus.COMPLETE, None, None + if from_stream_id is None: + return TaskStatus.COMPLETE, None, None - await self.clock.sleep(DeviceWriterHandler.DEVICE_MSGS_DELETE_SLEEP) + await self.clock.sleep(DeviceWriterHandler.DEVICE_MSGS_DELETE_SLEEP) class DeviceWriterHandler(DeviceHandler): diff --git a/synapse/rest/synapse/mas/devices.py b/synapse/rest/synapse/mas/devices.py index b5d19d3cecc..9c56b2d6c44 100644 --- a/synapse/rest/synapse/mas/devices.py +++ b/synapse/rest/synapse/mas/devices.py @@ -62,17 +62,19 @@ async def _async_render_POST( body = parse_and_validate_json_object_from_request(request, self.PostBody) user_id = UserID(body.localpart, self.hostname) - - # Check the user exists - user = await self.store.get_user_by_id(user_id=str(user_id)) - if user is None: - raise NotFoundError("User not found") - - inserted = await self.device_handler.upsert_device( - user_id=str(user_id), - device_id=body.device_id, - display_name=body.display_name, - ) + user_id_str = str(user_id) + + async with self.device_handler.device_management_linearizer.queue(user_id_str): + # Check the user exists + user = await self.store.get_user_by_id(user_id=user_id_str) + if user is None: + raise NotFoundError("User not found") + + inserted = await self.device_handler.upsert_device( + user_id=user_id_str, + device_id=body.device_id, + display_name=body.display_name, + ) return HTTPStatus.CREATED if inserted else HTTPStatus.OK, {} @@ -103,16 +105,18 @@ async def _async_render_POST( body = parse_and_validate_json_object_from_request(request, self.PostBody) user_id = UserID(body.localpart, self.hostname) + user_id_str = str(user_id) - # Check the user exists - user = await self.store.get_user_by_id(user_id=str(user_id)) - if user is None: - raise NotFoundError("User not found") + async with self.device_handler.device_management_linearizer.queue(user_id_str): + # Check the user exists + user = await self.store.get_user_by_id(user_id=user_id_str) + if user is None: + raise NotFoundError("User not found") - await self.device_handler.delete_devices( - user_id=str(user_id), - device_ids=[body.device_id], - ) + await self.device_handler.delete_devices( + user_id=user_id_str, + device_ids=[body.device_id], + ) return HTTPStatus.NO_CONTENT, {} @@ -144,17 +148,19 @@ async def _async_render_POST( body = parse_and_validate_json_object_from_request(request, self.PostBody) user_id = UserID(body.localpart, self.hostname) - - # Check the user exists - user = await self.store.get_user_by_id(user_id=str(user_id)) - if user is None: - raise NotFoundError("User not found") - - await self.device_handler.update_device( - user_id=str(user_id), - device_id=body.device_id, - content={"display_name": body.display_name}, - ) + user_id_str = str(user_id) + + async with self.device_handler.device_management_linearizer.queue(user_id_str): + # Check the user exists + user = await self.store.get_user_by_id(user_id=user_id_str) + if user is None: + raise NotFoundError("User not found") + + await self.device_handler.update_device( + user_id=user_id_str, + device_id=body.device_id, + content={"display_name": body.display_name}, + ) return HTTPStatus.OK, {} @@ -186,64 +192,66 @@ async def _async_render_POST( body = parse_and_validate_json_object_from_request(request, self.PostBody) user_id = UserID(body.localpart, self.hostname) - - # Check the user exists - user = await self.store.get_user_by_id(user_id=str(user_id)) - if user is None: - raise NotFoundError("User not found") - - current_devices = await self.store.get_devices_by_user(user_id=str(user_id)) - current_devices_list = set(current_devices.keys()) - target_device_list = set(body.devices) - - # Exclude the dehydrated device (MSC3814): it has no MAS session, so MAS - # never lists it in the target set and the reconciliation below would - # otherwise treat it as extra and delete it. This mirrors the admin - # devices API and MAS's own legacy device-sync path, which both skip it. - dehydrated_device = await self.device_handler.get_dehydrated_device( - user_id=str(user_id) - ) - if dehydrated_device is not None: - current_devices_list.discard(dehydrated_device[0]) - - to_add = target_device_list - current_devices_list - to_delete = current_devices_list - target_device_list - - # Log what we're about to do to make it easier to debug if it stops - # mid-way, as this can be a long operation if there are a lot of devices - # to delete or to add. - if to_add and to_delete: - logger.info( - "Syncing %d devices for user %s will add %d devices and delete %d devices", - len(target_device_list), - user_id, - len(to_add), - len(to_delete), - ) - elif to_add: - logger.info( - "Syncing %d devices for user %s will add %d devices", - len(target_device_list), - user_id, - len(to_add), - ) - elif to_delete: - logger.info( - "Syncing %d devices for user %s will delete %d devices", - len(target_device_list), - user_id, - len(to_delete), - ) - - if to_delete: - await self.device_handler.delete_devices( - user_id=str(user_id), device_ids=to_delete - ) - - for device_id in to_add: - await self.device_handler.upsert_device( - user_id=str(user_id), - device_id=device_id, + user_id_str = str(user_id) + + async with self.device_handler.device_management_linearizer.queue(user_id_str): + # Check the user exists + user = await self.store.get_user_by_id(user_id=user_id_str) + if user is None: + raise NotFoundError("User not found") + + current_devices = await self.store.get_devices_by_user(user_id=user_id_str) + current_devices_list = set(current_devices.keys()) + target_device_list = set(body.devices) + + # Exclude the dehydrated device (MSC3814): it has no MAS session, so MAS + # never lists it in the target set and the reconciliation below would + # otherwise treat it as extra and delete it. This mirrors the admin + # devices API and MAS's own legacy device-sync path, which both skip it. + dehydrated_device = await self.device_handler.get_dehydrated_device( + user_id=user_id_str ) + if dehydrated_device is not None: + current_devices_list.discard(dehydrated_device[0]) + + to_add = target_device_list - current_devices_list + to_delete = current_devices_list - target_device_list + + # Log what we're about to do to make it easier to debug if it stops + # mid-way, as this can be a long operation if there are a lot of devices + # to delete or to add. + if to_add and to_delete: + logger.info( + "Syncing %d devices for user %s will add %d devices and delete %d devices", + len(target_device_list), + user_id, + len(to_add), + len(to_delete), + ) + elif to_add: + logger.info( + "Syncing %d devices for user %s will add %d devices", + len(target_device_list), + user_id, + len(to_add), + ) + elif to_delete: + logger.info( + "Syncing %d devices for user %s will delete %d devices", + len(target_device_list), + user_id, + len(to_delete), + ) + + if to_delete: + await self.device_handler.delete_devices( + user_id=user_id_str, device_ids=to_delete + ) + + for device_id in to_add: + await self.device_handler.upsert_device( + user_id=user_id_str, + device_id=device_id, + ) return 200, {}