mirror of
https://github.com/element-hq/synapse.git
synced 2026-08-25 07:30:17 +00:00
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.
This commit is contained in:
Andrew Morgan
+17
-13
@@ -139,6 +139,9 @@ class DeviceHandler:
|
||||
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 @@ class DeviceHandler:
|
||||
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):
|
||||
|
||||
@@ -62,17 +62,19 @@ class MasUpsertDeviceResource(MasBaseResource):
|
||||
|
||||
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")
|
||||
|
||||
inserted = await self.device_handler.upsert_device(
|
||||
user_id=str(user_id),
|
||||
device_id=body.device_id,
|
||||
display_name=body.display_name,
|
||||
)
|
||||
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 @@ class MasDeleteDeviceResource(MasBaseResource):
|
||||
|
||||
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 @@ class MasUpdateDeviceDisplayNameResource(MasBaseResource):
|
||||
|
||||
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.update_device(
|
||||
user_id=str(user_id),
|
||||
device_id=body.device_id,
|
||||
content={"display_name": body.display_name},
|
||||
)
|
||||
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 @@ class MasSyncDevicesResource(MasBaseResource):
|
||||
|
||||
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")
|
||||
|
||||
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)
|
||||
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=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),
|
||||
# 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])
|
||||
|
||||
if to_delete:
|
||||
await self.device_handler.delete_devices(
|
||||
user_id=str(user_id), device_ids=to_delete
|
||||
)
|
||||
to_add = target_device_list - current_devices_list
|
||||
to_delete = current_devices_list - target_device_list
|
||||
|
||||
for device_id in to_add:
|
||||
await self.device_handler.upsert_device(
|
||||
user_id=str(user_id),
|
||||
device_id=device_id,
|
||||
)
|
||||
# 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, {}
|
||||
|
||||
Reference in New Issue
Block a user