diff options
| -rw-r--r-- | changelog.d/16492.misc | 1 | ||||
| -rw-r--r-- | synapse/handlers/device.py | 2 | ||||
| -rw-r--r-- | synapse/storage/databases/main/deviceinbox.py | 15 |
3 files changed, 11 insertions, 7 deletions
diff --git a/changelog.d/16492.misc b/changelog.d/16492.misc new file mode 100644 index 0000000000..ecb3356bdd --- /dev/null +++ b/changelog.d/16492.misc @@ -0,0 +1 @@ +Improve performance of delete device messages query, cf issue [16479](https://github.com/matrix-org/synapse/issues/16479). diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py index 544bc7c13d..3ce96ef3cb 100644 --- a/synapse/handlers/device.py +++ b/synapse/handlers/device.py @@ -592,6 +592,8 @@ class DeviceHandler(DeviceWorkerHandler): ) # Delete device messages asynchronously and in batches using the task scheduler + # We specify an upper stream id to avoid deleting non delivered messages + # if an user re-uses a device ID. await self._task_scheduler.schedule_task( DELETE_DEVICE_MSGS_TASK_NAME, resource_id=device_id, diff --git a/synapse/storage/databases/main/deviceinbox.py b/synapse/storage/databases/main/deviceinbox.py index 1faa6f04b2..3e7425d4a6 100644 --- a/synapse/storage/databases/main/deviceinbox.py +++ b/synapse/storage/databases/main/deviceinbox.py @@ -478,18 +478,19 @@ class DeviceInboxWorkerStore(SQLBaseStore): log_kv({"message": "No changes in cache since last check"}) return 0 - ROW_ID_NAME = self.database_engine.row_id_name - def delete_messages_for_device_txn(txn: LoggingTransaction) -> int: limit_statement = "" if limit is None else f"LIMIT {limit}" sql = f""" - DELETE FROM device_inbox WHERE {ROW_ID_NAME} IN ( - SELECT {ROW_ID_NAME} FROM device_inbox - WHERE user_id = ? AND device_id = ? AND stream_id <= ? - {limit_statement} + DELETE FROM device_inbox WHERE user_id = ? AND device_id = ? AND stream_id <= ( + SELECT MAX(stream_id) FROM ( + SELECT stream_id FROM device_inbox + WHERE user_id = ? AND device_id = ? AND stream_id <= ? + ORDER BY stream_id + {limit_statement} + ) AS q1 ) """ - txn.execute(sql, (user_id, device_id, up_to_stream_id)) + txn.execute(sql, (user_id, device_id, user_id, device_id, up_to_stream_id)) return txn.rowcount count = await self.db_pool.runInteraction( |