From 578a5e24a905c5d90d5c609cb485a5ab7277f8a5 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 5 Jul 2022 13:51:04 +0100 Subject: [PATCH] Use upserts for updating `event_push_summary` (#13153) --- changelog.d/13153.misc | 1 + .../databases/main/event_push_actions.py | 47 +++---------------- 2 files changed, 8 insertions(+), 40 deletions(-) create mode 100644 changelog.d/13153.misc diff --git a/changelog.d/13153.misc b/changelog.d/13153.misc new file mode 100644 index 000000000..3bb51962e --- /dev/null +++ b/changelog.d/13153.misc @@ -0,0 +1 @@ +Reduce DB usage of `/sync` when a large number of unread messages have recently been sent in a room. diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index bb6e104d7..32536430a 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -1013,8 +1013,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas sql = """ SELECT user_id, room_id, coalesce(old.%s, 0) + upd.cnt, - upd.stream_ordering, - old.user_id + upd.stream_ordering FROM ( SELECT user_id, room_id, count(*) as cnt, max(stream_ordering) as stream_ordering @@ -1042,7 +1041,6 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas summaries[(row[0], row[1])] = _EventPushSummary( unread_count=row[2], stream_ordering=row[3], - old_user_id=row[4], notif_count=0, ) @@ -1063,57 +1061,27 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas summaries[(row[0], row[1])] = _EventPushSummary( unread_count=0, stream_ordering=row[3], - old_user_id=row[4], notif_count=row[2], ) logger.info("Rotating notifications, handling %d rows", len(summaries)) - # If the `old.user_id` above is NULL then we know there isn't already an - # entry in the table, so we simply insert it. Otherwise we update the - # existing table. - self.db_pool.simple_insert_many_txn( + self.db_pool.simple_upsert_many_txn( txn, table="event_push_summary", - keys=( - "user_id", - "room_id", - "notif_count", - "unread_count", - "stream_ordering", - ), - values=[ + key_names=("user_id", "room_id"), + key_values=[(user_id, room_id) for user_id, room_id in summaries], + value_names=("notif_count", "unread_count", "stream_ordering"), + value_values=[ ( - user_id, - room_id, summary.notif_count, summary.unread_count, summary.stream_ordering, ) - for ((user_id, room_id), summary) in summaries.items() - if summary.old_user_id is None + for summary in summaries.values() ], ) - txn.execute_batch( - """ - UPDATE event_push_summary - SET notif_count = ?, unread_count = ?, stream_ordering = ? - WHERE user_id = ? AND room_id = ? - """, - ( - ( - summary.notif_count, - summary.unread_count, - summary.stream_ordering, - user_id, - room_id, - ) - for ((user_id, room_id), summary) in summaries.items() - if summary.old_user_id is not None - ), - ) - txn.execute( "UPDATE event_push_summary_stream_ordering SET stream_ordering = ?", (rotate_to_stream_ordering,), @@ -1293,5 +1261,4 @@ class _EventPushSummary: unread_count: int stream_ordering: int - old_user_id: str notif_count: int