Skip to content

Add a linearizer to device deletion requests #19965

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
anoadragon453 wants to merge 1 commit into
base: develop
Choose a base branch
from
+113 −101
Draft

Add a linearizer to device deletion requests #19965

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 17 additions & 13 deletions synapse/handlers/device.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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):
Expand Down
184 changes: 96 additions & 88 deletions synapse/rest/synapse/mas/devices.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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, {}

Expand Down Expand Up @@ -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, {}

Expand Down Expand Up @@ -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, {}

Expand Down Expand Up @@ -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, {}
Loading