MirrorHub/synapse
0
0
Fork 1
mirror of https://mau.dev/maunium/synapse.git synced 2024-06-18 10:38:21 +02:00
Code Issues Projects Releases Wiki Activity

Limit the number of events sent over replication when persisting events. (#10082)

Browse source
This commit is contained in:
Brendan Abolivier 2021-05-27 18:10:58 +02:00 committed by GitHub
parent 8e132fe64e
commit f828a70be3
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 11 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
changelog.d/10082.bugfix Normal file
View file

@ -0,0 +1 @@
Fixed a bug causing replication requests to fail when receiving a lot of events via federation.

17
synapse/handlers/federation.py
View file

@ -91,6 +91,7 @@ from synapse.types import (
get_domain_from_id, get_domain_from_id,
) )
from synapse.util.async_helpers import Linearizer, concurrently_execute from synapse.util.async_helpers import Linearizer, concurrently_execute
from synapse.util.iterutils import batch_iter
from synapse.util.retryutils import NotRetryingDestination from synapse.util.retryutils import NotRetryingDestination
from synapse.util.stringutils import shortstr from synapse.util.stringutils import shortstr
from synapse.visibility import filter_events_for_server from synapse.visibility import filter_events_for_server
@ -3053,13 +3054,15 @@ class FederationHandler(BaseHandler):
""" """
instance = self.config.worker.events_shard_config.get_instance(room_id) instance = self.config.worker.events_shard_config.get_instance(room_id)
if instance != self._instance_name: if instance != self._instance_name:
result = await self._send_events( # Limit the number of events sent over federation.
instance_name=instance, for batch in batch_iter(event_and_contexts, 1000):
store=self.store, result = await self._send_events(
room_id=room_id, instance_name=instance,
event_and_contexts=event_and_contexts, store=self.store,
backfilled=backfilled, room_id=room_id,
) event_and_contexts=batch,
backfilled=backfilled,
)
return result["max_stream_id"] return result["max_stream_id"]
else: else:
assert self.storage.persistence assert self.storage.persistence