From 130fd45393d39425a4aa35a388b1e5a74ba6f6f4 Mon Sep 17 00:00:00 2001 From: Brendan Abolivier Date: Wed, 16 Feb 2022 12:16:48 +0100 Subject: [PATCH] Limit concurrent AS joins (#11996) Initially introduced in matrix-org-hotfixes by e5537cf (and tweaked by later commits). Fixes #11995 See also #4826 --- changelog.d/11996.misc | 1 + synapse/handlers/room_member.py | 46 +++++++++++++++++++-------------- 2 files changed, 28 insertions(+), 19 deletions(-) create mode 100644 changelog.d/11996.misc diff --git a/changelog.d/11996.misc b/changelog.d/11996.misc new file mode 100644 index 000000000000..6c675fd1931c --- /dev/null +++ b/changelog.d/11996.misc @@ -0,0 +1 @@ +Limit concurrent joins from applications services. \ No newline at end of file diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py index bf1a47efb020..b2adc0f48be9 100644 --- a/synapse/handlers/room_member.py +++ b/synapse/handlers/room_member.py @@ -82,6 +82,7 @@ def __init__(self, hs: "HomeServer"): self.event_auth_handler = hs.get_event_auth_handler() self.member_linearizer: Linearizer = Linearizer(name="member") + self.member_as_limiter = Linearizer(max_count=10, name="member_as_limiter") self.clock = hs.get_clock() self.spam_checker = hs.get_spam_checker() @@ -500,25 +501,32 @@ async def update_membership( key = (room_id,) - with (await self.member_linearizer.queue(key)): - result = await self.update_membership_locked( - requester, - target, - room_id, - action, - txn_id=txn_id, - remote_room_hosts=remote_room_hosts, - third_party_signed=third_party_signed, - ratelimit=ratelimit, - content=content, - new_room=new_room, - require_consent=require_consent, - outlier=outlier, - historical=historical, - allow_no_prev_events=allow_no_prev_events, - prev_event_ids=prev_event_ids, - auth_event_ids=auth_event_ids, - ) + as_id = object() + if requester.app_service: + as_id = requester.app_service.id + + # We first linearise by the application service (to try to limit concurrent joins + # by application services), and then by room ID. + with (await self.member_as_limiter.queue(as_id)): + with (await self.member_linearizer.queue(key)): + result = await self.update_membership_locked( + requester, + target, + room_id, + action, + txn_id=txn_id, + remote_room_hosts=remote_room_hosts, + third_party_signed=third_party_signed, + ratelimit=ratelimit, + content=content, + new_room=new_room, + require_consent=require_consent, + outlier=outlier, + historical=historical, + allow_no_prev_events=allow_no_prev_events, + prev_event_ids=prev_event_ids, + auth_event_ids=auth_event_ids, + ) return result