dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-28 09:24:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Optimize mark_as_sent_devices_by_remote to do less CPU work on the database (#20120)

Browse Source 
### 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))

<img width="919" height="265" alt="CPU of database server"
src="https://github.com/user-attachments/assets/9a43cf3c-bf9a-4ade-b70e-0bf0fb53f0ef"
/>


@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))

<img width="919" height="269" alt="'mark_as_sent_devices_by_remote'
spiking in the 'DB transactions by total txn time' graph"
src="https://github.com/user-attachments/assets/9336ce29-c045-4940-9c46-9e4906300f2d"
/>

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.
This commit is contained in:
Eric Eastwood
2026-08-18 09:27:50 +01:00
committed by GitHub
parent 94a5f2afb3
commit ba3ab1876a
2 changed files with 40 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changelog.d/20120.misc
+1
View File
@@ -0,0 +1 @@
Reduce database CPU usage when marking device list changes as sent over federation.
synapse/storage/databases/main/devices.py
+39 -16
View File
@@ -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: