diff options
Diffstat (limited to 'synapse/federation/sender')
| -rw-r--r-- | synapse/federation/sender/__init__.py | 10 | ||||
| -rw-r--r-- | synapse/federation/sender/per_destination_queue.py | 12 | ||||
| -rw-r--r-- | synapse/federation/sender/transaction_manager.py | 5 |
3 files changed, 18 insertions, 9 deletions
diff --git a/synapse/federation/sender/__init__.py b/synapse/federation/sender/__init__.py index 643b26ae6d..97fc4d0a82 100644 --- a/synapse/federation/sender/__init__.py +++ b/synapse/federation/sender/__init__.py @@ -328,7 +328,9 @@ class FederationSender: # to allow us to perform catch-up later on if the remote is unreachable # for a while. await self.store.store_destination_rooms_entries( - destinations, pdu.room_id, pdu.internal_metadata.stream_ordering, + destinations, + pdu.room_id, + pdu.internal_metadata.stream_ordering, ) for destination in destinations: @@ -475,7 +477,7 @@ class FederationSender: self, states: List[UserPresenceState], destinations: List[str] ) -> None: """Send the given presence states to the given destinations. - destinations (list[str]) + destinations (list[str]) """ if not states or not self.hs.config.use_presence: @@ -616,8 +618,8 @@ class FederationSender: last_processed = None # type: Optional[str] while True: - destinations_to_wake = await self.store.get_catch_up_outstanding_destinations( - last_processed + destinations_to_wake = ( + await self.store.get_catch_up_outstanding_destinations(last_processed) ) if not destinations_to_wake: diff --git a/synapse/federation/sender/per_destination_queue.py b/synapse/federation/sender/per_destination_queue.py index db8e456fe8..deb519f3ef 100644 --- a/synapse/federation/sender/per_destination_queue.py +++ b/synapse/federation/sender/per_destination_queue.py @@ -85,7 +85,8 @@ class PerDestinationQueue: # processing. We have a guard in `attempt_new_transaction` that # ensure we don't start sending stuff. logger.error( - "Create a per destination queue for %s on wrong worker", destination, + "Create a per destination queue for %s on wrong worker", + destination, ) self._should_send_on_this_instance = False @@ -440,8 +441,10 @@ class PerDestinationQueue: if first_catch_up_check: # first catchup so get last_successful_stream_ordering from database - self._last_successful_stream_ordering = await self._store.get_destination_last_successful_stream_ordering( - self._destination + self._last_successful_stream_ordering = ( + await self._store.get_destination_last_successful_stream_ordering( + self._destination + ) ) if self._last_successful_stream_ordering is None: @@ -457,7 +460,8 @@ class PerDestinationQueue: # get at most 50 catchup room/PDUs while True: event_ids = await self._store.get_catch_up_room_event_ids( - self._destination, self._last_successful_stream_ordering, + self._destination, + self._last_successful_stream_ordering, ) if not event_ids: diff --git a/synapse/federation/sender/transaction_manager.py b/synapse/federation/sender/transaction_manager.py index 3e07f925e0..763aff296c 100644 --- a/synapse/federation/sender/transaction_manager.py +++ b/synapse/federation/sender/transaction_manager.py @@ -65,7 +65,10 @@ class TransactionManager: @measure_func("_send_new_transaction") async def send_new_transaction( - self, destination: str, pdus: List[EventBase], edus: List[Edu], + self, + destination: str, + pdus: List[EventBase], + edus: List[Edu], ) -> bool: """ Args: |