From ba3ab1876acd326fa5cc9b161e87b064ac54f3ed Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Tue, 18 Aug 2026 03:27:50 -0500 Subject: [PATCH] Optimize `mark_as_sent_devices_by_remote` to do less CPU work on the database (#20120) ### Background On `matrix.org`, since 2026-08-15, we are seeing the database CPU being saturated more than usual ([grafana](https://grafana.matrix.org/d/rYdddlPWk/node-exporter?orgId=1&from=2026-08-10T23:23:14.237Z&to=2026-08-17T23:23:14.237Z&timezone=browser&var-DS_PROMETHEUS=default&var-job=machine&var-node=matrix-db-01.matrix.org:9100&var-diskdevices=%5Ba-z%5D%2B%7Cnvme%5B0-9%5D%2Bn%5B0-9%5D%2B&refresh=1m&viewPanel=panel-77)) CPU of database server @reivilibre [found](https://matrix.to/#/!yHWhpxlXVaLcsgDUKb:matrix.org/$fUMf60IbFocpEuJNbmbEzFiCyKCFnAl4I4XpB27DUQs?via=banzan.uk&via=element.io&via=matrix.org) `mark_as_sent_devices_by_remote` spiking in the `DB transactions by total txn time` graph ([grafana](https://grafana.matrix.org/d/000000012/synapse?var-bucket_size=$__auto&orgId=1&from=2026-08-10T22:26:07.336Z&to=2026-08-17T22:26:07.336Z&timezone=browser&var-datasource=default&var-instance=matrix.org&var-job=synapse_federation_sender&var-index=$__all&showCategory=Thresholds&viewPanel=panel-11)) 'mark_as_sent_devices_by_remote' spiking in the 'DB transactions by total txn time' graph And we indeed see a bunch of time being spent on `mark_as_sent_devices_by_remote` by looking at `pg_stat_statements`. The top two queries we're spending CPU on are the `SELECT` and `DELETE` statements in [`mark_as_sent_devices_by_remote`](https://github.com/element-hq/synapse/blob/94a5f2afb36a4afdc36813fd1c24b9a1c4aec252/synapse/storage/databases/main/devices.py#L922-L950). We also did some related work in this area recently with https://github.com/element-hq/synapse/pull/20098 although it was tackling a different bottle-neck. ### This PR This PRs combines the two separate `SELECT` and `DELETE` queries which touch the same data into one `DELETE ... RETURNING ...` query. this means we get to save the cost of one of those statements (less CPU on the database) and fewer statements per transaction (less round-trips) means connections turn over faster, and can move on to process the next thing. This is a micro-optimization I spotted while reading [`_mark_as_sent_devices_by_remote_txn`](https://github.com/element-hq/synapse/blob/develop/synapse/storage/databases/main/devices.py#L922-L950) rather than a structural fix. I'm sure there are even better things to do to where we could even avoid this kind of work altogether but this seemed like a quick win especially given how much this particular transaction is saturating the database. Doing things faster doesn't necessarily mean we solve the saturated/starved CPU problem but it does mean the same task costs less CPU. From the `pg_stat_statements` samples above, we can expect to be up to ~27% more efficient with database CPU on this code path (derived from the `cpus_busy` numbers above `22.932 / (22.932 + 61.956)`). We're removing the `SELECT` (`61.956` `cpus_busy`) but I still expect the `DELETE` (`22.932` `cpus_busy`) to take on a similar cost as I'm sure it's a warm cache situation between them. --- changelog.d/20120.misc | 1 + synapse/storage/databases/main/devices.py | 55 ++++++++++++++++------- 2 files changed, 40 insertions(+), 16 deletions(-) create mode 100644 changelog.d/20120.misc diff --git a/changelog.d/20120.misc b/changelog.d/20120.misc new file mode 100644 index 0000000000..d59db030c6 --- /dev/null +++ b/changelog.d/20120.misc @@ -0,0 +1 @@ +Reduce database CPU usage when marking device list changes as sent over federation. diff --git a/synapse/storage/databases/main/devices.py b/synapse/storage/databases/main/devices.py index 69470ae6e6..6cf9270ff2 100644 --- a/synapse/storage/databases/main/devices.py +++ b/synapse/storage/databases/main/devices.py @@ -922,33 +922,56 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore): def _mark_as_sent_devices_by_remote_txn( self, txn: LoggingTransaction, destination: str, stream_id: int ) -> None: - # We update the device_lists_outbound_last_success with the successfully - # poked users. + # Delete all sent outbound pokes, returning them so that we can update + # `device_lists_outbound_last_success` with the successfully poked users. + # + # This is a high frequency transaction (runs very often when processing a + # backlog of device list changes) and can bog down the database CPU with the + # sheer number of statements. + # + # We prefer to trade a little bit of processing time on the Python side + # (aggregating `max_stream_id_by_user_id`) as the alternative would be to have + # two separate queries; a `SELECT ... GROUP BY user_id` with the aggregation and + # then a `DELETE` which means we touch the same rows twice. We get to save the + # cost of one of those statements (less CPU on the database) and fewer + # statements per transaction (less round-trips) means connections turn over + # faster, and can move on to process the next thing. + # + # By the nature of `MAX_EDUS_PER_TRANSACTION`, we're only dealing with 100 rows + # at max which is pretty trivial for us to process on the Python side. sql = """ - SELECT user_id, coalesce(max(o.stream_id), 0) - FROM device_lists_outbound_pokes as o - WHERE destination = ? AND o.stream_id <= ? - GROUP BY user_id + DELETE FROM device_lists_outbound_pokes + WHERE destination = ? AND stream_id <= ? + RETURNING user_id, stream_id """ txn.execute(sql, (destination, stream_id)) - rows = txn.fetchall() + # Aggregate `max_stream_id_by_user_id` + max_stream_id_by_user_id: dict[str, int] = {} + for user_id, poke_stream_id in txn: + max_stream_id_by_user_id[user_id] = max( + max_stream_id_by_user_id.get(user_id, 0), poke_stream_id + ) + + # Update `device_lists_outbound_last_success` with the successfully poked + # users. + # + # We could potentially combine this in one big CTE with the query above but it + # isn't supported by SQLite (SQLite doesn't support `DELETE` in a CTE). self.db_pool.simple_upsert_many_txn( txn=txn, table="device_lists_outbound_last_success", key_names=("destination", "user_id"), - key_values=[(destination, user_id) for user_id, _ in rows], + key_values=[ + (destination, user_id) for user_id in max_stream_id_by_user_id.keys() + ], value_names=("stream_id",), - value_values=[(stream_id,) for _, stream_id in rows], + value_values=[ + (user_stream_id,) + for user_stream_id in max_stream_id_by_user_id.values() + ], ) - # Delete all sent outbound pokes - sql = """ - DELETE FROM device_lists_outbound_pokes - WHERE destination = ? AND stream_id <= ? - """ - txn.execute(sql, (destination, stream_id)) - async def add_user_signature_change_to_streams( self, from_user_id: str, user_ids: list[str] ) -> int: