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, {}